KAFKA-7699: Anchored punctuation (#19937)

This PR implements Processor API interfaces for using Anchored
punctuation. As described in

https://cwiki.apache.org/confluence/display/KAFKA/KIP-1146%3A+Anchored+punctuation,
anchored punctuations allows the users of the PAPI to set both a start
time and an interval for when the punctuations should be triggered. The
changes have been based on all the places where

`org.apache.kafka.streams.processor.internals.ProcessorContext.schedule()`
is used.

The changes consist of introducing new interfaces as the punctuation
time logic in the
`org.apache.kafka.streams.processor.internals.PunctuationSchedule`
already supports calculating a new punctuation time based on a start
time and an interval.

The original `schedule()` implementations have been refactored into
using the new `schedule()` implementation supporting the `startTime`
parameter. For the original implementations, the `startTime` parameter
has been sat to `null`, effectively using the "method overloading"
programming technique.

Reviewers: Bill Bejeck<[email protected]>

Author: hermanjakobsen

streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java

@@ -30,6 +30,7 @@
 
 import java.io.File;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
 
 /**
@@ -157,6 +158,46 @@ Cancellable schedule(final Duration interval,
                          final PunctuationType type,
                          final Punctuator callback);
 
+    /**
+     * Schedule a periodic operation for processors. A processor may call this method during a
+     * {@link org.apache.kafka.streams.kstream.KTable#transformValues(ValueTransformerWithKeySupplier, String...)}'s
+     * {@link org.apache.kafka.streams.kstream.ValueTransformerWithKey#init(ProcessorContext) initialization} or
+     * {@link org.apache.kafka.streams.kstream.ValueTransformerWithKey#transform(Object, Object) processing} to
+     * schedule a periodic callback — called a punctuation — to {@link Punctuator#punctuate(long)}.
+     * The type parameter controls what notion of time is used for punctuation:
+     * <ul>
+     *   <li>{@link PunctuationType#STREAM_TIME} &mdash; uses "stream time", which is advanced by the processing of messages
+     *   in accordance with the timestamp as extracted by the {@link TimestampExtractor} in use.
+     *   The first punctuation will be triggered by the first record that is processed.
+     *   <b>NOTE:</b> Only advances as messages arrive</li>
+     *   <li>{@link PunctuationType#WALL_CLOCK_TIME} &mdash; uses system time (the wall-clock time),
+     *   which advances independent of whether new messages arrive.
+     *   The first punctuation will be triggered after interval has elapsed.
+     *   <b>NOTE:</b> This is best effort only as its granularity is limited by how long an iteration of the
+     *   processing loop takes to complete</li>
+     * </ul>
+     *
+     * <b>Skipping punctuations:</b> Punctuations will not be triggered more than once at any given timestamp.
+     * This means that "missed" punctuation will be skipped.
+     * It's possible to "miss" a punctuation if:
+     * <ul>
+     *   <li>with {@link PunctuationType#STREAM_TIME}, when stream time advances more than interval</li>
+     *   <li>with {@link PunctuationType#WALL_CLOCK_TIME}, on GC pause, too short interval, ...</li>
+     * </ul>
+     *
+     * @param startTime the time for the first punctuation.
+     *                  The subsequent trigger times are calculated using the {@code startTime} and the {@code interval}
+     * @param interval the time interval between punctuations (supported minimum is 1 millisecond)
+     * @param type one of: {@link PunctuationType#STREAM_TIME}, {@link PunctuationType#WALL_CLOCK_TIME}
+     * @param callback a function consuming timestamps representing the current stream or system time
+     * @return a handle allowing cancellation of the punctuation schedule established by this method
+     * @throws IllegalArgumentException if the interval is not representable in milliseconds
+     */
+    Cancellable schedule(final Instant startTime,
+                         final Duration interval,
+                         final PunctuationType type,
+                         final Punctuator callback);
+
     /**
      * Forward a key/value pair to all downstream processors.
      * Used the input record's timestamp as timestamp for the output record.

streams/src/main/java/org/apache/kafka/streams/processor/api/ProcessingContext.java

@@ -32,6 +32,7 @@
 
 import java.io.File;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
 import java.util.Optional;
 
@@ -159,6 +160,46 @@ Cancellable schedule(final Duration interval,
                          final PunctuationType type,
                          final Punctuator callback);
 
+    /**
+     * Schedule a periodic operation for processors. A processor may call this method during
+     * {@link Processor#init(ProcessorContext) initialization},
+     * {@link Processor#process(Record) processing},
+     * {@link FixedKeyProcessor#init(FixedKeyProcessorContext) initialization}, or
+     * {@link FixedKeyProcessor#process(FixedKeyRecord) processing} to
+     * schedule a periodic callback — called a punctuation — to {@link Punctuator#punctuate(long)}.
+     * The type parameter controls what notion of time is used for punctuation:
+     * <ul>
+     *   <li>{@link PunctuationType#STREAM_TIME} &mdash; uses "stream time", which is advanced by the processing of messages
+     *   in accordance with the timestamp as extracted by the {@link TimestampExtractor} in use.
+     *   The first punctuation will be triggered by the first record that is processed.
+     *   <b>NOTE:</b> Only advances as messages arrive</li>
+     *   <li>{@link PunctuationType#WALL_CLOCK_TIME} &mdash; uses system time (the wall-clock time),
+     *   which advances independent of whether new messages arrive.
+     *   The first punctuation will be triggered after interval has elapsed.
+     *   <b>NOTE:</b> This is best effort only as its granularity is limited by how long an iteration of the
+     *   processing loop takes to complete</li>
+     * </ul>
+     *
+     * <b>Skipping punctuations:</b> Punctuations will not be triggered more than once at any given timestamp.
+     * This means that "missed" punctuation will be skipped.
+     * It's possible to "miss" a punctuation if:
+     * <ul>
+     *   <li>with {@link PunctuationType#STREAM_TIME}, when stream time advances more than interval</li>
+     *   <li>with {@link PunctuationType#WALL_CLOCK_TIME}, on GC pause, too short interval, ...</li>
+     * </ul>
+     * @param startTime the time for the first punctuation. The subsequent trigger times are calculated
+     *                  using the {@code startTime} and the {@code interval}
+     * @param interval the time interval between punctuations (supported minimum is 1 millisecond)
+     * @param type one of: {@link PunctuationType#STREAM_TIME}, {@link PunctuationType#WALL_CLOCK_TIME}
+     * @param callback a function consuming timestamps representing the current stream or system time
+     * @return a handle allowing cancellation of the punctuation schedule established by this method
+     * @throws IllegalArgumentException if the interval is not representable in milliseconds
+     */
+    Cancellable schedule(final Instant startTime,
+                         final Duration interval,
+                         final PunctuationType type,
+                         final Punctuator callback);
+
     /**
      * Request a commit. Note that calling {@code commit()} is only a request for a commit, but it does not execute one.
      * Hence, when {@code commit()} returns, no commit was executed yet. However, Kafka Streams will commit as soon

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

@@ -31,6 +31,7 @@
 
 import java.io.File;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
 import java.util.Objects;
 
@@ -96,6 +97,14 @@ public Cancellable schedule(final Duration interval,
         return delegate.schedule(interval, type, callback);
     }
 
+    @Override
+    public Cancellable schedule(final Instant startTime,
+                                final Duration interval,
+                                final PunctuationType type,
+                                final Punctuator callback) {
+        return delegate.schedule(startTime, interval, type, callback);
+    }
+
     @Override
     public <K, V> void forward(final K key, final V value) {
         throw new StreamsException(EXPLANATION);

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

@@ -33,6 +33,7 @@
 import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
 
 import java.time.Duration;
+import java.time.Instant;
 
 import static org.apache.kafka.streams.processor.internals.AbstractReadWriteDecorator.wrapWithReadWriteStore;
 
@@ -133,6 +134,11 @@ public Cancellable schedule(final Duration interval, final PunctuationType type,
         throw new UnsupportedOperationException("this should not happen: schedule() not supported in global processor context.");
     }
 
+    @Override
+    public Cancellable schedule(final Instant startTime, final Duration interval, final PunctuationType type, final Punctuator callback) {
+        throw new UnsupportedOperationException("this should not happen: schedule() not supported in global processor context.");
+    }
+
     @Override
     public void logChange(final String storeName,
                           final Bytes key,

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

@@ -39,6 +39,7 @@
 import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntryFlushListener;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -313,7 +314,24 @@ public Cancellable schedule(final Duration interval,
         if (intervalMs < 1) {
             throw new IllegalArgumentException("The minimum supported scheduling interval is 1 millisecond.");
         }
-        return streamTask.schedule(intervalMs, type, callback);
+        return streamTask.schedule(intervalMs, type, callback);    }
+
+    @Override
+    public Cancellable schedule(
+            final Instant startTime,
+            final Duration interval,
+            final PunctuationType type,
+            final Punctuator callback) throws IllegalArgumentException {
+        throwUnsupportedOperationExceptionIfStandby("schedule");
+        final String msgPrefix = prepareMillisCheckFailMsgPrefix(interval, "interval");
+        final long intervalMs = validateMillisecondDuration(interval, msgPrefix);
+        if (intervalMs < 1) {
+            throw new IllegalArgumentException("The minimum supported scheduling interval is 1 millisecond.");
+        }
+        if (startTime.isBefore(Instant.EPOCH)) {
+            throw new IllegalArgumentException("The minimum supported start time is Instant.EPOCH.");
+        }
+        return streamTask.schedule(startTime, intervalMs, type, callback);
     }
 
     @Override

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

@@ -69,7 +69,7 @@ boolean isCancelled() {
     public PunctuationSchedule next(final long currTimestamp) {
         long nextPunctuationTime = timestamp + interval;
         if (currTimestamp >= nextPunctuationTime) {
-            // we missed one ore more punctuations
+            // we missed one or more punctuations
             // avoid scheduling a new punctuations immediately, this can happen:
             // - when using STREAM_TIME punctuation and there was a gap i.e., no data was
             //   received for at least 2*interval

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

@@ -53,6 +53,7 @@
 import org.apache.kafka.streams.state.internals.ThreadCache;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -1185,15 +1186,27 @@ public Cancellable schedule(final long interval, final PunctuationType type, fin
         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);
         }
     }
 
+    /**
+     * Schedules a punctuation for the processor
+     *
+     * @param startTime time of the first punctuation
+     * @param interval the interval in milliseconds
+     * @param type     the punctuation type
+     * @throws IllegalStateException if the current node is not null
+     */
+    public Cancellable schedule(final Instant startTime, final long interval, final PunctuationType type, final Punctuator punctuator) {
+        return schedule(startTime.toEpochMilli(), interval, type, punctuator, true);
+    }
+
     /**
      * Schedules a punctuation for the processor
      *
@@ -1202,12 +1215,12 @@ public Cancellable schedule(final long interval, final PunctuationType type, fin
      * @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:
@@ -1222,6 +1235,15 @@ private Cancellable schedule(final long startTime, final long interval, final Pu
         }
     }
 
+    // For anchored schedule, we want to have all punctuations only fire 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/AbstractProcessorContextTest.java

@@ -44,6 +44,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Properties;
 
 import static org.apache.kafka.test.StreamsTestUtils.getStreamsConfig;
@@ -221,6 +222,14 @@ public Cancellable schedule(final Duration interval,
             return null;
         }
 
+        @Override
+        public Cancellable schedule(final Instant startTime,
+                                    final Duration interval,
+                                    final PunctuationType type,
+                                    final Punctuator callback) {
+            return null;
+        }
+
         @Override
         public <K, V> void forward(final Record<K, V> record) {}