fix: ensure that first anchored punctuation happens at a combination of startTime + interval

Author: hermanjakobsen

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -1150,15 +1150,15 @@ public Cancellable schedule(final Instant startTime, final long interval, final
             switch (type) {
                 case STREAM_TIME:
                     // align punctuation to 0L, punctuate as soon as we have data
-                    return schedule(0L, interval, type, punctuator);
+                    return schedule(0L, interval, type, punctuator, false);
                 case WALL_CLOCK_TIME:
                     // align punctuation to now, punctuate after interval has elapsed
-                    return schedule(time.milliseconds() + interval, interval, type, punctuator);
+                    return schedule(time.milliseconds() + interval, interval, type, punctuator, false);
                 default:
                     throw new IllegalArgumentException("Unrecognized PunctuationType: " + type);
             }
         }
-        return schedule(startTime.toEpochMilli(), interval, type, punctuator);
+        return schedule(startTime.toEpochMilli(), interval, type, punctuator, true);
     }
 
     /**
@@ -1169,12 +1169,12 @@ public Cancellable schedule(final Instant startTime, final long interval, final
      * @param type      the punctuation type
      * @throws IllegalStateException if the current node is not null
      */
-    private Cancellable schedule(final long startTime, final long interval, final PunctuationType type, final Punctuator punctuator) {
+    private Cancellable schedule(final long startTime, final long interval, final PunctuationType type, final Punctuator punctuator, final boolean anchored) {
         if (processorContext.currentNode() == null) {
             throw new IllegalStateException(String.format("%sCurrent node is null", logPrefix));
         }
 
-        final PunctuationSchedule schedule = new PunctuationSchedule(processorContext.currentNode(), startTime, interval, punctuator);
+        final PunctuationSchedule schedule = getInitialSchedule(startTime, interval, type, punctuator, anchored);
 
         switch (type) {
             case STREAM_TIME:
@@ -1189,6 +1189,15 @@ private Cancellable schedule(final long startTime, final long interval, final Pu
         }
     }
 
+    // For anchored schedule, we want to have all punctuations to happen only on times based on combinations of startTime and interval
+    // This method ensures that the first anchored punctuation is not fired prematurely due to startTime < now
+    private PunctuationSchedule getInitialSchedule(final long startTime, final long interval, final PunctuationType type, final Punctuator punctuator, final boolean anchored) {
+        final PunctuationSchedule originalSchedule = new PunctuationSchedule(processorContext.currentNode(), startTime, interval, punctuator);
+        final long now = (type == PunctuationType.WALL_CLOCK_TIME) ? time.milliseconds() : streamTime();
+
+        return (anchored && startTime < now) ? originalSchedule.next(now) : originalSchedule;
+    }
+
     /**
      * Possibly trigger registered stream-time punctuation functions if
      * current partition group timestamp has reached the defined stamp

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java

@@ -1760,10 +1760,8 @@ public void shouldPunctuateUsingAnchoredSystemStartTimeWithStartTimeBeforeNow()
         task.initializeIfNeeded();
         task.completeRestoration(noOpResetter -> { });
 
-        // now is after startTime -> initial punctuation
-        assertTrue(task.canPunctuateSystemTime());
-        assertTrue(task.maybePunctuateSystemTime());
-
+        // Expects the punctuations to happen at a time satisfying startTime + n * interval
+        // where n is a positive number or 0
         time.sleep(9);
         assertFalse(task.canPunctuateSystemTime());
         assertFalse(task.maybePunctuateSystemTime());
@@ -1773,7 +1771,7 @@ public void shouldPunctuateUsingAnchoredSystemStartTimeWithStartTimeBeforeNow()
         time.sleep(10);
         assertTrue(task.canPunctuateSystemTime());
         assertTrue(task.maybePunctuateSystemTime());
-        anchoredProcessorSystemTime.mockProcessor.checkAndClearPunctuateResult(PunctuationType.WALL_CLOCK_TIME, testStartTime, testStartTime + 10, testStartTime + 20);
+        anchoredProcessorSystemTime.mockProcessor.checkAndClearPunctuateResult(PunctuationType.WALL_CLOCK_TIME,  testStartTime + 10, testStartTime + 20);
     }
 
     @Test