You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/25 20:02:10 UTC
[GitHub] [kafka] vvcephei opened a new pull request #11952: MINOR: Fix stream-join metadata
vvcephei opened a new pull request #11952:
URL: https://github.com/apache/kafka/pull/11952
https://github.com/apache/kafka/pull/11356 inadvertently changed
the (undefined) header forwarding behavior of stream-stream joins.
This change does not define the behavior, but just restores the prior
undefined behavior for continuity's sake. Defining the header-forwarding
behavior is future work.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#issuecomment-1079392675
ping @jeqo , in case you have a chance to review as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#discussion_r835654066
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java
##########
@@ -887,6 +892,98 @@ public void shouldNotEmitLeftJoinResultForAsymmetricAfterWindow() {
}
}
+ /**
+ * NOTE: Header forwarding is undefined behavior, but we still want to understand the
+ * behavior so that we can make decisions about defining it in the future.
+ */
+ @Test
+ public void shouldForwardCurrentHeaders() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KStream<Integer, String> stream1;
+ final KStream<Integer, String> stream2;
+ final KStream<Integer, String> joined;
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
+ stream1 = builder.stream(topic1, consumed);
+ stream2 = builder.stream(topic2, consumed);
+
+ joined = stream1.outerJoin(
+ stream2,
+ MockValueJoiner.TOSTRING_JOINER,
+ JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(10L)),
+ StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
+ );
+ joined.process(supplier);
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), PROPS)) {
+ final TestInputTopic<Integer, String> inputTopic1 =
+ driver.createInputTopic(topic1, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
+ final TestInputTopic<Integer, String> inputTopic2 =
+ driver.createInputTopic(topic2, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
+ final MockApiProcessor<Integer, String, Void, Void> processor = supplier.theCapturedProcessor();
+
+ inputTopic1.pipeInput(new TestRecord<>(
+ 0,
+ "A0",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}),
+ 0L
+ ));
+ inputTopic2.pipeInput(new TestRecord<>(
+ 1,
+ "a0",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}),
+ 0L
+ ));
+ // bump stream-time to trigger outer-join results
+ inputTopic2.pipeInput(new TestRecord<>(
+ 3,
+ "dummy",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}),
+ (long) 211
+ ));
+
+ // Again, header forwarding is undefined, but the current observed behavior is that
+ // the headers pass through the forwarding record.
+ processor.checkAndClearProcessedRecords(
+ new Record<>(
+ 1,
+ "null+a0",
+ 0L,
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})})
Review comment:
Semantically really bad to forward `0x3`, but well, it is what it is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#discussion_r835654720
##########
File path: streams/src/test/java/org/apache/kafka/test/MockApiProcessor.java
##########
@@ -92,6 +93,20 @@ public void process(final Record<KIn, VIn> record) {
}
public void checkAndClearProcessResult(final KeyValueTimestamp<?, ?>... expected) {
Review comment:
Seems we should get rid of this one after we have updated the full DSL to use the new `Record` type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#issuecomment-1080887880
Thanks for the quick reviews, @mjsax and @jeqo !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#discussion_r836631366
##########
File path: streams/src/test/java/org/apache/kafka/test/MockApiProcessor.java
##########
@@ -92,6 +93,20 @@ public void process(final Record<KIn, VIn> record) {
}
public void checkAndClearProcessResult(final KeyValueTimestamp<?, ?>... expected) {
Review comment:
We certainly could, but there's also nothing semantically wrong with making KVT-based assertions, so I figured I'd just leave them both in the test infrastructure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#discussion_r836629808
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java
##########
@@ -887,6 +892,98 @@ public void shouldNotEmitLeftJoinResultForAsymmetricAfterWindow() {
}
}
+ /**
+ * NOTE: Header forwarding is undefined behavior, but we still want to understand the
+ * behavior so that we can make decisions about defining it in the future.
+ */
+ @Test
+ public void shouldForwardCurrentHeaders() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KStream<Integer, String> stream1;
+ final KStream<Integer, String> stream2;
+ final KStream<Integer, String> joined;
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
+ stream1 = builder.stream(topic1, consumed);
+ stream2 = builder.stream(topic2, consumed);
+
+ joined = stream1.outerJoin(
+ stream2,
+ MockValueJoiner.TOSTRING_JOINER,
+ JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(10L)),
+ StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
+ );
+ joined.process(supplier);
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), PROPS)) {
+ final TestInputTopic<Integer, String> inputTopic1 =
+ driver.createInputTopic(topic1, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
+ final TestInputTopic<Integer, String> inputTopic2 =
+ driver.createInputTopic(topic2, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
+ final MockApiProcessor<Integer, String, Void, Void> processor = supplier.theCapturedProcessor();
+
+ inputTopic1.pipeInput(new TestRecord<>(
+ 0,
+ "A0",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x1})}),
+ 0L
+ ));
+ inputTopic2.pipeInput(new TestRecord<>(
+ 1,
+ "a0",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x2})}),
+ 0L
+ ));
+ // bump stream-time to trigger outer-join results
+ inputTopic2.pipeInput(new TestRecord<>(
+ 3,
+ "dummy",
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})}),
+ (long) 211
+ ));
+
+ // Again, header forwarding is undefined, but the current observed behavior is that
+ // the headers pass through the forwarding record.
+ processor.checkAndClearProcessedRecords(
+ new Record<>(
+ 1,
+ "null+a0",
+ 0L,
+ new RecordHeaders(new Header[]{new RecordHeader("h", new byte[]{0x3})})
Review comment:
Yeah, I was a bit bummed to see this, but at least now we know!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11952:
URL: https://github.com/apache/kafka/pull/11952#issuecomment-1080880253
It seems like the build is getting better. Only a couple of unrelated test failures:
```
[Build / JDK 8 and Scala 2.12 / kafka.server.ListOffsetsRequestTest.testCurrentEpochValidation()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11952/1/testReport/junit/kafka.server/ListOffsetsRequestTest/Build___JDK_8_and_Scala_2_12___testCurrentEpochValidation__/)
Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.integration.StreamsUncaughtExceptionHandlerIntegrationTest.shouldShutdownMultipleThreadApplication
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #11952: MINOR: Fix stream-join metadata
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #11952:
URL: https://github.com/apache/kafka/pull/11952
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org