lct45 commented on a change in pull request #9039:
URL: https://github.com/apache/kafka/pull/9039#discussion_r465094963
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
##########
@@ -459,6 +460,95 @@ public void shouldGroupByKey() throws Exception {
)));
}
+
+
+ @Test
+ public void shouldReduceSlidingWindows() throws Exception {
+ streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 0);
+ final long firstBatchTimestamp = 2000L;
+ final long timeDifference = 1000L;
+ produceMessages(firstBatchTimestamp);
+ final long secondBatchTimestamp = firstBatchTimestamp + timeDifference
/ 2;
+ produceMessages(secondBatchTimestamp);
+ final long thirdBatchTimestamp = secondBatchTimestamp + timeDifference
- 100L;
+ produceMessages(thirdBatchTimestamp);
+
+ final Serde<Windowed<String>> windowedSerde =
WindowedSerdes.timeWindowedSerdeFrom(String.class);
+ groupedStream
+
.windowedBy(SlidingWindows.withTimeDifferenceAndGrace(ofMillis(timeDifference),
ofMillis(2000L)))
+ .reduce(reducer)
+ .toStream()
+ .to(outputTopic, Produced.with(windowedSerde,
Serdes.String()));
+
+ startStreams();
+
+ final List<KeyValueTimestamp<Windowed<String>, String>> windowedOutput
= receiveMessages(
+ new TimeWindowedDeserializer<>(),
+ new StringDeserializer(),
+ String.class,
+ 25);
+
+ // read from ConsoleConsumer
+ final String resultFromConsoleConsumer =
readWindowedKeyedMessagesViaConsoleConsumer(
+ new TimeWindowedDeserializer<String>(),
+ new StringDeserializer(),
+ String.class,
+ 25,
+ true);
+
+ final Comparator<KeyValueTimestamp<Windowed<String>, String>>
comparator =
+ Comparator.comparing((KeyValueTimestamp<Windowed<String>,
String> o) -> o.key().key())
+ .thenComparing(KeyValueTimestamp::value);
+
+ windowedOutput.sort(comparator);
+ final long firstBatchLeftWindow = firstBatchTimestamp - timeDifference;
+ final long firstBatchRightWindow = firstBatchTimestamp + 1;
+ final long secondBatchLeftWindow = secondBatchTimestamp -
timeDifference;
+ final long secondBatchRightWindow = secondBatchTimestamp + 1;
+ final long thirdBatchLeftWindow = thirdBatchTimestamp - timeDifference;
+
+ final List<KeyValueTimestamp<Windowed<String>, String>> expectResult =
Arrays.asList(
+ new KeyValueTimestamp<>(new Windowed<>("A", new
TimeWindow(firstBatchLeftWindow, Long.MAX_VALUE)), "A", firstBatchTimestamp),
Review comment:
Because the windows are sorted, the windows created by each record
aren't consecutive, so I added comments describing each window, but only did it
for A since all the other keys are processed the exact same way. Sample
comment: `// A @ secondBatchTimestamp left window created when A @
secondBatchTimestamp processed`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
