[GitHub] [kafka] mjsax commented on a change in pull request #10824: KAFKA-12718 [WIP]: SessionWindows are closed too early

[GitBox]

mjsax commented on a change in pull request #10824:
URL: https://github.com/apache/kafka/pull/10824#discussion_r654149249



##########
File path: 
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
##########
@@ -421,8 +421,8 @@ public void 
shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace() {
         context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));
         processor.process("OnTime1", "1");
 
-        // dummy record to advance stream time = 1
-        context.setRecordContext(new ProcessorRecordContext(1, -2, -3, 
"topic", new RecordHeaders()));
+        // dummy record to advance stream time = 11
+        context.setRecordContext(new ProcessorRecordContext(10 + 1, -2, -3, 
"topic", new RecordHeaders()));

Review comment:
       Why `10 + 1`? Just make it `11`. Use the comment to explain the why

##########
File path: 
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
##########
@@ -421,8 +421,8 @@ public void 
shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace() {
         context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));

Review comment:
       Wondering if we should bump this timestamp to 10?

##########
File path: 
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
##########
@@ -421,8 +421,8 @@ public void 
shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace() {
         context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));

Review comment:
       Wondering if we should bump this timestamp to 10? (Or course, we would 
need to bump 11 below to 21 for this case)

##########
File path: 
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
##########
@@ -489,16 +489,16 @@ public void 
shouldLogAndMeterWhenSkippingLateRecordWithNonzeroGrace() {
             context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));
             processor.process("OnTime1", "1");
 
-            // dummy record to advance stream time = 1
-            context.setRecordContext(new ProcessorRecordContext(1, -2, -3, 
"topic", new RecordHeaders()));
+            // dummy record to advance stream time = 11
+            context.setRecordContext(new ProcessorRecordContext(10 + 1, -2, 
-3, "topic", new RecordHeaders()));

Review comment:
       as above

##########
File path: 
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
##########
@@ -489,16 +489,16 @@ public void 
shouldLogAndMeterWhenSkippingLateRecordWithNonzeroGrace() {
             context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));
             processor.process("OnTime1", "1");
 
-            // dummy record to advance stream time = 1
-            context.setRecordContext(new ProcessorRecordContext(1, -2, -3, 
"topic", new RecordHeaders()));
+            // dummy record to advance stream time = 11
+            context.setRecordContext(new ProcessorRecordContext(10 + 1, -2, 
-3, "topic", new RecordHeaders()));
             processor.process("dummy", "dummy");
 
             // delayed record arrives on time, should not be skipped
             context.setRecordContext(new ProcessorRecordContext(0, -2, -3, 
"topic", new RecordHeaders()));
             processor.process("OnTime2", "1");
 
-            // dummy record to advance stream time = 2
-            context.setRecordContext(new ProcessorRecordContext(2, -2, -3, 
"topic", new RecordHeaders()));
+            // dummy record to advance stream time = 12
+            context.setRecordContext(new ProcessorRecordContext(10 + 2, -2, 
-3, "topic", new RecordHeaders()));

Review comment:
       as above




-- 
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