VictorvandenHoven commented on code in PR #15510:
URL: https://github.com/apache/kafka/pull/15510#discussion_r1528270350
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java:
##########
@@ -511,14 +511,88 @@ public void testGracePeriod() {
// w2 = { 0:a0 (ts: 101), 1:a1 (ts: 101) }
// --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 0) }
// --> w2 = { 0:a0 (ts: 101), 1:a1 (ts: 101), 0:dummy (ts: 112) }
- inputTopic2.pipeInput(0, "dummy", 211);
+ inputTopic2.pipeInput(0, "dummy", 112);
processor.checkAndClearProcessResult(
new KeyValueTimestamp<>(1, "null+a1", 0L),
new KeyValueTimestamp<>(0, "A0+null", 0L)
);
}
}
+ @Test
+ public void testEmitAllNonJoinedResultsForAsymmetricWindow() {
+ 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.ofTimeDifferenceWithNoGrace(ofMillis(5)).after(ofMillis(20)),
+ 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();
+
+ // push one item to the primary stream; this should not produce
any items because there are no joins
+ // and window has not ended
+ // w1 = {}
+ // w2 = {}
+ // --> w1 = { 0:A0 (ts: 29) }
+ // --> w2 = {}
+ inputTopic1.pipeInput(0, "A0", 29L);
+ processor.checkAndClearProcessResult();
+
+ // push another item to the primary stream; this should not
produce any items because there are no joins
+ // and window has not ended
+ // w1 = { 0:A0 (ts: 29) }
+ // w2 = {}
+ // --> w1 = { 0:A0 (ts: 29), 1:A1 (ts: 30) }
+ // --> w2 = {}
+ inputTopic1.pipeInput(1, "A1", 30L);
+ processor.checkAndClearProcessResult();
+
+ // push one item to the other stream; this should not produce any
items because there are no joins
+ // and window has not ended
+ // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 30) }
+ // w2 = {}
+ // --> w1 = { 0:A0 (ts: 29), 1:A1 (ts: 30) }
+ // --> w2 = { 2:a2 (ts: 31) }
+ inputTopic2.pipeInput(2, "a2", 31L);
+ processor.checkAndClearProcessResult();
+
Review Comment:
Good idea, added a step with right hand side record on ts=36
--
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]
