VictorvandenHoven commented on code in PR #14426:
URL: https://github.com/apache/kafka/pull/14426#discussion_r1358134854
##########
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) }
+ // w2 = {}
+ final long time = 1000L;
+ for (int i = 0; i < expectedKeys.length; i++) {
+ inputTopic1.pipeInput(expectedKeys[i], "B" + expectedKeys[i],
time + i);
+ }
+ processor.checkAndClearProcessResult(
+ new KeyValueTimestamp<>(0, "B0+null", 1000L),
+ new KeyValueTimestamp<>(1, "B1+null", 1001L),
+ new KeyValueTimestamp<>(2, "B2+null", 1002L)
+ );
+ }
Review Comment:
No,it is a leftjoin with joinwindows 100ms before and 0ms after.
So directly at timestamp of B0, B1 and B2 the not-joined record will be
emitted.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
