[GitHub] [kafka] wcarlson5 commented on a change in pull request #9253: KAFKA-10366 & KAFKA-9649: Implement KIP-659 to allow TimeWindowedDeserializer and TimeWindowedSerde to handle window size

[GitBox <gi...@apache.org>]

wcarlson5 commented on a change in pull request #9253:
URL: https://github.com/apache/kafka/pull/9253#discussion_r562150019



##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
##########
@@ -235,25 +235,27 @@ public void shouldReduceWindowed() throws Exception {
                 .thenComparing(KeyValueTimestamp::value);
 
         windowedOutput.sort(comparator);
-        final long firstBatchWindow = firstBatchTimestamp / 500 * 500;
-        final long secondBatchWindow = secondBatchTimestamp / 500 * 500;
+        final long firstBatchWindowStart = firstBatchTimestamp / 500 * 500;
+        final long firstBatchWindowEnd = firstBatchWindowStart + 500;

Review comment:
       Why don't you use `timeDifference` like you did below? 

##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializer.java
##########
@@ -33,22 +33,22 @@
  */
 public class TimeWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
 
-    private final Long windowSize;
+    private Long windowSize;
     private boolean isChangelogTopic;
 
     private Deserializer<T> inner;
-    
+
     // Default constructor needed by Kafka
     public TimeWindowedDeserializer() {
-        this(null, Long.MAX_VALUE);
+        this(null, null);
     }
 
-    // TODO: fix this part as last bits of KAFKA-4468
+    @Deprecated
     public TimeWindowedDeserializer(final Deserializer<T> inner) {
         this(inner, Long.MAX_VALUE);

Review comment:
       Do you want to change this from `Long.MAX_VALUE` as well?

##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
##########
@@ -327,7 +329,7 @@ public void shouldAggregateWindowed() throws Exception {
         startStreams();
 
         final List<KeyValueTimestamp<Windowed<String>, Integer>> windowedMessages = receiveMessagesWithTimestamp(
-            new TimeWindowedDeserializer<>(),

Review comment:
       I don't think we should get rid of the generics unless we have to

##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializerTest.java
##########
@@ -56,4 +57,23 @@ public void testWindowedValueDeserializerNoArgConstructors() {
         assertNotNull("Inner deserializer should be not null", inner);
         assertTrue("Inner deserializer type should be ByteArrayDeserializer", inner instanceof ByteArrayDeserializer);
     }
+
+    @Test
+    public void setWindowSizeThroughConfigs() {
+        props.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, "500");
+        final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
+        deserializer.configure(props, false);

Review comment:
       Is there some sort of check you can verify here?




----------------------------------------------------------------
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:
users@infra.apache.org