mjsax commented on a change in pull request #9039:
URL: https://github.com/apache/kafka/pull/9039#discussion_r466028866
##########
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);
Review comment:
I guess testing both cases would be good. Even if testing the former
(fall outside of all existing windows) was my original intent.
And thank for comment Sophie: I tend to forget that we should produce all
(non-empty) right windows already upfront/eagerly (and not delayed/lazily when
stream-time advances beyond window-end time). In any case, it seems to be a
good test case to make sure we don't (re-)emit an (unexpected) window if
stream-time jumps ahead?
----------------------------------------------------------------
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
