mjsax commented on a change in pull request #9039: URL: https://github.com/apache/kafka/pull/9039#discussion_r466026007
########## 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: The ticket is already fixed. You need to pass in the `windowSize` into the the constructor of `TimeWindowDeserializer` to get rid of the problem. ---------------------------------------------------------------- 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