mjsax commented on code in PR #14426: URL: https://github.com/apache/kafka/pull/14426#discussion_r1511998434
########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java: ########## @@ -438,13 +438,13 @@ public void testOrdering() { inputTopic1.pipeInput(1, "A1", 100L); processor.checkAndClearProcessResult(); - // push one item to the other window that has a join; this should produce non-joined records with a closed window first, then - // the joined records - // by the time they were produced before + // push one item to the other window that has a join; + // this should produce the joined record first; + // then the not-joined record Review Comment: This change to the comment needs to be rolled back, right? We indeed produce the left-null join result first. ########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java: ########## @@ -884,11 +886,13 @@ public void shouldNotEmitLeftJoinResultForAsymmetricAfterWindow() { processor.checkAndClearProcessResult(); - // push one item to the first stream; this should produce one full-join item + // push one item to the first stream; + // this should produce one inner-join item; + // and a right-joined item for a3 Review Comment: We don't produce output for `a3` ########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java: ########## @@ -436,6 +436,239 @@ public void testRightNonJoinedRecordsAreNeverEmittedByTheRightProcessor() { } } + @Test + public void testLeftNonJoinedRecordsWithZeroAfterAreEmitted() { + final StreamsBuilder builder = new StreamsBuilder(); + + final int[] expectedKeys = new int[] {0, 1, 2, 3}; + + 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.leftJoin( + stream2, + MockValueJoiner.TOSTRING_JOINER, + JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)).after(ZERO), + StreamJoined.with(Serdes.Integer(), + Serdes.String(), + Serdes.String()) + ); + joined.process(supplier); + + final Collection<Set<String>> copartitionGroups = + TopologyWrapper.getInternalTopologyBuilder(builder.build()).copartitionGroups(); + + assertEquals(1, copartitionGroups.size()); + assertEquals(new HashSet<>(Arrays.asList(topic1, topic2)), copartitionGroups.iterator().next()); + + 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(); + + processor.init(null); + // push four items with increasing timestamps to the primary stream; this should emit null-joined items + // w1 = {} + // w2 = {} + // --> w1 = { 0:B0 (ts: 1000), 1:B1 (ts: 1001), 2:B2 (ts: 1002), 3:B3 (ts: 1003) } Review Comment: Why are we using `B` not `A` for left input? ########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java: ########## @@ -436,6 +436,239 @@ public void testRightNonJoinedRecordsAreNeverEmittedByTheRightProcessor() { } } + @Test + public void testLeftNonJoinedRecordsWithZeroAfterAreEmitted() { Review Comment: Why do we need this test case? Seems it's fully contained in `testLeftJoinedRecordsWithZeroAfterAreEmitted` below? ########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java: ########## @@ -436,6 +436,239 @@ public void testRightNonJoinedRecordsAreNeverEmittedByTheRightProcessor() { } } + @Test + public void testLeftNonJoinedRecordsWithZeroAfterAreEmitted() { + final StreamsBuilder builder = new StreamsBuilder(); + + final int[] expectedKeys = new int[] {0, 1, 2, 3}; + + 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.leftJoin( + stream2, + MockValueJoiner.TOSTRING_JOINER, + JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)).after(ZERO), + StreamJoined.with(Serdes.Integer(), + Serdes.String(), + Serdes.String()) + ); + joined.process(supplier); + + final Collection<Set<String>> copartitionGroups = + TopologyWrapper.getInternalTopologyBuilder(builder.build()).copartitionGroups(); + + assertEquals(1, copartitionGroups.size()); + assertEquals(new HashSet<>(Arrays.asList(topic1, topic2)), copartitionGroups.iterator().next()); + + 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(); + + processor.init(null); + // push four items with increasing timestamps to the primary stream; this should emit null-joined items Review Comment: `his should emit [three] null-joined items; the last one does not emit yet` -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org