lct45 commented on a change in pull request #9039:
URL: https://github.com/apache/kafka/pull/9039#discussion_r466070317
##########
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:
Created a ticket for this here:
https://issues.apache.org/jira/browse/KAFKA-10366 let me know if the
description isn't clear
----------------------------------------------------------------
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
