This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 695ccca9a04 [FLINK-40101][runtime] Emit intermediate watermarks while
firing timers
695ccca9a04 is described below
commit 695ccca9a04a11275ffb230024527e3f226b9ab8
Author: Piotr Nowojski <[email protected]>
AuthorDate: Wed Jul 8 18:05:49 2026 +0200
[FLINK-40101][runtime] Emit intermediate watermarks while firing timers
With unaligned checkpoints + interruptible timers, an operator's output
watermark could stall for hours (surviving restarts) because it only
advances once an entire due-timer backlog drains in one uninterrupted
pass — a large backlog (e.g. after a rescale) can outlast every single
attempt.
InternalTimerServiceImpl/InternalTimeServiceManagerImpl now track the
highest watermark known to be fully fired even when interrupted
partway, and MailboxWatermarkProcessor emits that as an intermediate
watermark instead of withholding all progress. This progress lives in
a new field, not currentWatermark, since currentWatermark's eager
semantics are relied on elsewhere (WindowOperator cleanup timers, user
ProcessFunctions). Emission is paced by a configurable interval
(default 5s, 0 disables) via an internal no-op processing-time nudge,
avoiding per-timer clock checks.
---
.../generated/checkpointing_configuration.html | 6 +
.../flink/configuration/CheckpointingOptions.java | 25 +++
.../api/operators/AbstractStreamOperator.java | 17 +-
.../api/operators/AbstractStreamOperatorV2.java | 17 +-
.../api/operators/InternalTimeServiceManager.java | 15 ++
.../operators/InternalTimeServiceManagerImpl.java | 60 +++++++-
.../api/operators/InternalTimerServiceImpl.java | 21 +++
.../api/operators/MailboxWatermarkProcessor.java | 21 ++-
.../BatchExecutionInternalTimeServiceManager.java | 8 +
.../operators/MailboxWatermarkProcessorTest.java | 89 ++++++++++-
...nalignedCheckpointsInterruptibleTimersTest.java | 171 +++++++++++++++++++++
.../join/interval/RowTimeIntervalJoinTest.java | 3 +
...oInputStreamOperatorWithStateRetentionTest.java | 4 +
...essSetTableOperatorInterruptibleTimersTest.java | 7 +-
.../sort/BaseTemporalSortOperatorTest.java | 6 +
.../eventtime/EventTimeWatermarkHandlerTest.java | 5 +
16 files changed, 464 insertions(+), 11 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
index 704ac492aa8..2fcbc5fc1ed 100644
--- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
@@ -176,6 +176,12 @@
<td>Boolean</td>
<td>Forces unaligned checkpoints, particularly allowing them for
iterative jobs.</td>
</tr>
+ <tr>
+
<td><h5>execution.checkpointing.unaligned.interruptible-timers.emit-intermediate-watermarks</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>When unaligned checkpoints with interruptible timers are
enabled (see 'execution.checkpointing.unaligned.interruptible-timers.enabled')
and firing the timers due for a watermark advance is interrupted before
completing, an intermediate watermark reflecting the progress made so far is
emitted downstream, at most as often as configured by
'pipeline.auto-watermark-interval'. This keeps downstream operators from
stalling on watermark progress during a long-running catch-up. S [...]
+ </tr>
<tr>
<td><h5>execution.checkpointing.unaligned.interruptible-timers.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
index fb653120944..75abd4178a0 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
@@ -611,6 +611,31 @@ public class CheckpointingOptions {
+ " For this feature to be enabled, it
must be also supported by the operator."
+ " Currently this is supported by all
TableStreamOperators and CepOperator.");
+ /**
+ * Controls whether an intermediate watermark is emitted while a watermark
advance is
+ * interrupted before completing, for unaligned checkpoints with
interruptible timers enabled
+ * (see {@link #ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS}). Has no effect
unless interruptible
+ * timers are enabled. The emission interval is governed by {@link
+ * PipelineOptions#AUTO_WATERMARK_INTERVAL}, the same as regular periodic
watermark generation,
+ * since both have comparable performance implications.
+ */
+ @Experimental
+ public static final ConfigOption<Boolean>
+ UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS =
+ ConfigOptions.key(
+
"execution.checkpointing.unaligned.interruptible-timers.emit-intermediate-watermarks")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "When unaligned checkpoints with
interruptible timers are enabled (see"
+ + "
'execution.checkpointing.unaligned.interruptible-timers.enabled') and"
+ + " firing the timers due for a
watermark advance is interrupted before"
+ + " completing, an intermediate
watermark reflecting the progress made so"
+ + " far is emitted downstream, at
most as often as configured by"
+ + "
'pipeline.auto-watermark-interval'. This keeps downstream operators"
+ + " from stalling on watermark
progress during a long-running catch-up."
+ + " Set to false to disable
intermediate watermark emission.");
+
public static final ConfigOption<Boolean>
ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH =
ConfigOptions.key("execution.checkpointing.checkpoints-after-tasks-finish")
.booleanType()
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index 58ad00e7727..d091128bdb5 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -71,6 +71,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Locale;
@@ -395,9 +396,23 @@ public abstract class AbstractStreamOperator<OUT>
&& areInterruptibleTimersConfigured()
&& getTimeServiceManager().isPresent()) {
LOG.info("Interruptible timers enabled for {}",
getClass().getSimpleName());
+ InternalTimeServiceManager<?> timeServiceManager =
getTimeServiceManager().get();
+ boolean emitIntermediateWatermarks =
+ getContainingTask()
+ .getJobConfiguration()
+ .get(
+ CheckpointingOptions
+
.UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS);
+ if (emitIntermediateWatermarks) {
+ timeServiceManager.configureIntermediateWatermarkInterval(
+
Duration.ofMillis(getExecutionConfig().getAutoWatermarkInterval()));
+ }
this.watermarkProcessor =
new MailboxWatermarkProcessor(
- output, mailboxExecutor,
getTimeServiceManager().get());
+ output,
+ mailboxExecutor,
+ timeServiceManager,
+ emitIntermediateWatermarks);
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java
index b898f439c1b..8dfd584351a 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java
@@ -68,6 +68,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Locale;
import java.util.Optional;
@@ -249,9 +250,23 @@ public abstract class AbstractStreamOperatorV2<OUT>
&& areInterruptibleTimersConfigured()
&& getTimeServiceManager().isPresent()) {
LOG.info("Interruptible timers enabled for {}",
getClass().getSimpleName());
+ InternalTimeServiceManager<?> timeServiceManager =
getTimeServiceManager().get();
+ boolean emitIntermediateWatermarks =
+ runtimeContext
+ .getJobConfiguration()
+ .get(
+ CheckpointingOptions
+
.UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS);
+ if (emitIntermediateWatermarks) {
+ timeServiceManager.configureIntermediateWatermarkInterval(
+
Duration.ofMillis(getExecutionConfig().getAutoWatermarkInterval()));
+ }
watermarkProcessor =
new MailboxWatermarkProcessor(
- output, mailboxExecutor,
getTimeServiceManager().get());
+ output,
+ mailboxExecutor,
+ timeServiceManager,
+ emitIntermediateWatermarks);
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
index e24e879afae..dcc37bc7bed 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
@@ -30,6 +30,7 @@ import
org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
import java.io.Serializable;
+import java.time.Duration;
/**
* An entity keeping all the time-related services.
@@ -79,6 +80,20 @@ public interface InternalTimeServiceManager<K> {
boolean tryAdvanceWatermark(Watermark watermark, ShouldStopAdvancingFn
shouldStopAdvancingFn)
throws Exception;
+ /**
+ * Configures how often an intermediate watermark should be made available
(see {@link
+ * #getReachedWatermark()}) while a {@link #tryAdvanceWatermark} call is
interrupted before
+ * completing. A {@code interval} of {@link Duration#ZERO zero} disables
this. Implementations
+ * that do not support interrupted watermark advancement may ignore this.
+ */
+ default void configureIntermediateWatermarkInterval(Duration interval) {}
+
+ /**
+ * Returns the highest watermark for which all managed {@link
InternalTimerService timer
+ * services} are known to have fired all due timers.
+ */
+ long getReachedWatermark();
+
/**
* Snapshots the timers to raw keyed state.
*
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java
index c8510995cbd..2b7f31c060a 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java
@@ -44,8 +44,11 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.io.InputStream;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -81,6 +84,11 @@ public class InternalTimeServiceManagerImpl<K> implements
InternalTimeServiceMan
@Nullable AsyncExecutionController<K, ?> asyncExecutionController;
+ private long intermediateWatermarkIntervalMs = 0;
+ private boolean intermediateWatermarkNudgeScheduled = false;
+ private long reachedWatermark = Long.MIN_VALUE;
+ private int nextServiceStartIndex = 0;
+
private InternalTimeServiceManagerImpl(
TaskIOMetricGroup taskIOMetricGroup,
KeyGroupRange localKeyGroupRange,
@@ -215,15 +223,59 @@ public class InternalTimeServiceManagerImpl<K> implements
InternalTimeServiceMan
}
}
+ @Override
+ public void configureIntermediateWatermarkInterval(Duration interval) {
+ this.intermediateWatermarkIntervalMs = interval.toMillis();
+ }
+
@Override
public boolean tryAdvanceWatermark(
Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn)
throws Exception {
- for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) {
- if (!service.tryAdvanceWatermark(watermark.getTimestamp(),
shouldStopAdvancingFn)) {
- return false;
+ maybeScheduleIntermediateWatermarkNudge();
+ List<InternalTimerServiceImpl<K, ?>> services = new
ArrayList<>(timerServices.values());
+ boolean fullyAdvanced = true;
+ for (int i = 0; i < services.size() && fullyAdvanced; i++) {
+ // Rotate the starting service every call so that a
persistently-behind service can't
+ // permanently starve the ones after it in a fixed iteration
order: once one service
+ // is interrupted, stop attempting to fire on the remaining ones
this round, but still
+ // fold their (possibly stale, from an earlier round)
reachedWatermark into the min
+ // below.
+ InternalTimerServiceImpl<K, ?> service =
+ services.get((nextServiceStartIndex + i) %
services.size());
+ if (fullyAdvanced) {
+ fullyAdvanced =
+ service.tryAdvanceWatermark(
+ watermark.getTimestamp(),
shouldStopAdvancingFn);
}
}
- return true;
+ if (!services.isEmpty()) {
+ nextServiceStartIndex = (nextServiceStartIndex + 1) %
services.size();
+ }
+ long minReachedWatermark = Long.MAX_VALUE;
+ for (InternalTimerServiceImpl<?, ?> service : services) {
+ minReachedWatermark = Math.min(minReachedWatermark,
service.getReachedWatermark());
+ }
+ reachedWatermark = minReachedWatermark;
+ return fullyAdvanced;
+ }
+
+ @Override
+ public long getReachedWatermark() {
+ return reachedWatermark;
+ }
+
+ // A firing loop only yields once the mailbox has other mail waiting;
ordinary record/watermark
+ // traffic doesn't go through the mailbox, so a long, otherwise-idle drain
would never yield on
+ // its own. This periodic no-op mail forces a yield point at roughly the
configured interval,
+ // without adding any per-timer clock check to the firing loop itself.
+ private void maybeScheduleIntermediateWatermarkNudge() {
+ if (intermediateWatermarkIntervalMs > 0 &&
!intermediateWatermarkNudgeScheduled) {
+ intermediateWatermarkNudgeScheduled = true;
+ processingTimeService.scheduleWithFixedDelay(
+ timestamp -> {},
+ intermediateWatermarkIntervalMs,
+ intermediateWatermarkIntervalMs);
+ }
}
////////////////// Fault Tolerance Methods
///////////////////
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java
index e24e8c9a62d..6beb8c87814 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java
@@ -71,6 +71,14 @@ public class InternalTimerServiceImpl<K, N> implements
InternalTimerService<N> {
*/
protected long currentWatermark = Long.MIN_VALUE;
+ /**
+ * Unlike {@link #currentWatermark}, which is set to the requested target
watermark before any
+ * timer fires, this only advances after each timer has actually fired, so
{@link
+ * #getReachedWatermark()} can safely surface it downstream even while
{@link
+ * #tryAdvanceWatermark} is interrupted before reaching its requested
target.
+ */
+ private long reachedWatermark = Long.MIN_VALUE;
+
/**
* The one and only Future (if any) registered to execute the next {@link
Triggerable} action,
* when its (processing) time arrives.
@@ -229,6 +237,10 @@ public class InternalTimerServiceImpl<K, N> implements
InternalTimerService<N> {
this.currentWatermark = watermark;
}
+ long getReachedWatermark() {
+ return reachedWatermark;
+ }
+
@Override
public void registerProcessingTimeTimer(N namespace, long time) {
InternalTimer<K, N> oldHead = processingTimeTimersQueue.peek();
@@ -339,10 +351,19 @@ public class InternalTimerServiceImpl<K, N> implements
InternalTimerService<N> {
eventTimeTimersQueue.poll();
triggerTarget.onEventTime(timer);
taskIOMetricGroup.getNumFiredTimers().inc();
+ // Other timers due at exactly this timestamp may still be
unfired, so only claim
+ // progress strictly below it.
+ reachedWatermark = timer.getTimestamp() - 1;
// Check if we should stop advancing after at least one iteration
to guarantee progress
// and prevent a potential starvation.
interrupted = shouldStopAdvancingFn.test();
}
+ if (!interrupted) {
+ // The loop above ran to completion: every timer due at or before
`time` has fired (or
+ // none were due), so the full requested watermark has been
reached, not just the last
+ // fired timer's timestamp.
+ reachedWatermark = time;
+ }
return !interrupted;
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java
index fb498f65f07..69c58f11097 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java
@@ -44,6 +44,7 @@ public class MailboxWatermarkProcessor<OUT> {
private final Output<StreamRecord<OUT>> output;
private final MailboxExecutor mailboxExecutor;
private final InternalTimeServiceManager<?> internalTimeServiceManager;
+ private final boolean emitIntermediateWatermarks;
/**
* Flag to indicate whether a progress watermark is scheduled in the
mailbox. This is used to
@@ -53,13 +54,17 @@ public class MailboxWatermarkProcessor<OUT> {
private Watermark maxInputWatermark = Watermark.UNINITIALIZED;
+ private long lastEmittedIntermediateWatermark = Long.MIN_VALUE;
+
public MailboxWatermarkProcessor(
Output<StreamRecord<OUT>> output,
MailboxExecutor mailboxExecutor,
- InternalTimeServiceManager<?> internalTimeServiceManager) {
+ InternalTimeServiceManager<?> internalTimeServiceManager,
+ boolean emitIntermediateWatermarks) {
this.output = checkNotNull(output);
this.mailboxExecutor = checkNotNull(mailboxExecutor);
this.internalTimeServiceManager =
checkNotNull(internalTimeServiceManager);
+ this.emitIntermediateWatermarks = emitIntermediateWatermarks;
}
public void emitWatermarkInsideMailbox(Watermark mark) throws Exception {
@@ -73,8 +78,20 @@ public class MailboxWatermarkProcessor<OUT> {
if (internalTimeServiceManager.tryAdvanceWatermark(
maxInputWatermark, mailboxExecutor::shouldInterrupt)) {
// In case output watermark has fully progressed emit it
downstream.
+ lastEmittedIntermediateWatermark =
maxInputWatermark.getTimestamp();
output.emitWatermark(maxInputWatermark);
- } else if (!progressWatermarkScheduled) {
+ return;
+ }
+ if (emitIntermediateWatermarks) {
+ long reachedWatermark =
internalTimeServiceManager.getReachedWatermark();
+ if (reachedWatermark > lastEmittedIntermediateWatermark) {
+ // Firing was interrupted before completing; surface the
progress made so far
+ // instead of leaving watermark advancement stalled until the
whole drain finishes.
+ lastEmittedIntermediateWatermark = reachedWatermark;
+ output.emitWatermark(new Watermark(reachedWatermark));
+ }
+ }
+ if (!progressWatermarkScheduled) {
progressWatermarkScheduled = true;
// We still have work to do, but we need to let other mails to be
processed first.
mailboxExecutor.execute(
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java
index df215a56501..aaaefc26dec 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java
@@ -56,6 +56,8 @@ public class BatchExecutionInternalTimeServiceManager<K>
// should perform correctly when the timer fires.
private final boolean asyncStateProcessingMode;
+ private long reachedWatermark = Long.MIN_VALUE;
+
public BatchExecutionInternalTimeServiceManager(
ProcessingTimeService processingTimeService, boolean
asyncStateProcessingMode) {
this.processingTimeService = checkNotNull(processingTimeService);
@@ -89,6 +91,7 @@ public class BatchExecutionInternalTimeServiceManager<K>
if (watermark.getTimestamp() == Long.MAX_VALUE) {
keySelected(null);
}
+ reachedWatermark = watermark.getTimestamp();
}
@Override
@@ -98,6 +101,11 @@ public class BatchExecutionInternalTimeServiceManager<K>
return true;
}
+ @Override
+ public long getReachedWatermark() {
+ return reachedWatermark;
+ }
+
@Override
public void snapshotToRawKeyedState(
KeyedStateCheckpointOutputStream context, String operatorName)
throws Exception {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java
b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java
index e0b76c4123f..299c511f166 100644
---
a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java
@@ -53,7 +53,8 @@ class MailboxWatermarkProcessorTest {
new CollectorOutput<>(emittedElements),
new MailboxExecutorImpl(
mailbox, priority,
StreamTaskActionExecutor.IMMEDIATE),
- timerService);
+ timerService,
+ true);
final List<Watermark> expectedOutput = new ArrayList<>();
watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(1));
watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(2));
@@ -84,6 +85,87 @@ class MailboxWatermarkProcessorTest {
assertThat(emittedElements).containsExactlyElementsOf(expectedOutput);
}
+ /**
+ * An interruption unrelated to the intermediate-watermark nudge (e.g. a
checkpoint mail) must
+ * not surface an intermediate watermark when the feature is disabled.
+ */
+ @Test
+ void testIntermediateWatermarkNotEmittedWhenDisabled() throws Exception {
+ int priority = 42;
+ final List<StreamElement> emittedElements = new ArrayList<>();
+ final TaskMailboxImpl mailbox = new TaskMailboxImpl();
+ final InternalTimeServiceManager<?> timerService =
+ new NoOpInternalTimeServiceManager() {
+ @Override
+ public long getReachedWatermark() {
+ return 5;
+ }
+ };
+
+ final MailboxWatermarkProcessor<StreamRecord<String>>
watermarkProcessor =
+ new MailboxWatermarkProcessor<>(
+ new CollectorOutput<>(emittedElements),
+ new MailboxExecutorImpl(
+ mailbox, priority,
StreamTaskActionExecutor.IMMEDIATE),
+ timerService,
+ false);
+
+ // Unrelated mail interrupts the firing loop for a reason unrelated to
the nudge.
+ mailbox.put(new Mail(() -> {}, TaskMailbox.MIN_PRIORITY, "checkpoint
mail"));
+
+ watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(10));
+
+ // configureIntermediateWatermarkInterval() was never called on this
manager.
+ assertThat(emittedElements).isEmpty();
+ }
+
+ /**
+ * Once a watermark has been fully emitted via the shortcut branch, a
later interrupted advance
+ * must not emit an intermediate watermark below it.
+ */
+ @Test
+ void testIntermediateWatermarkNeverBelowAlreadyEmittedWatermark() throws
Exception {
+ int priority = 42;
+ final List<StreamElement> emittedElements = new ArrayList<>();
+ final TaskMailboxImpl mailbox = new TaskMailboxImpl();
+ final boolean[] fullyAdvancedOnce = new boolean[] {false};
+ final InternalTimeServiceManager<?> timerService =
+ new NoOpInternalTimeServiceManager() {
+ @Override
+ public boolean tryAdvanceWatermark(
+ Watermark watermark, ShouldStopAdvancingFn
shouldStopAdvancingFn) {
+ if (!fullyAdvancedOnce[0]) {
+ fullyAdvancedOnce[0] = true;
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public long getReachedWatermark() {
+ return 5;
+ }
+ };
+
+ final MailboxWatermarkProcessor<StreamRecord<String>>
watermarkProcessor =
+ new MailboxWatermarkProcessor<>(
+ new CollectorOutput<>(emittedElements),
+ new MailboxExecutorImpl(
+ mailbox, priority,
StreamTaskActionExecutor.IMMEDIATE),
+ timerService,
+ true);
+
+ // Fully advances to 10 via the shortcut branch.
+ watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(10));
+ // A new, higher watermark arrives but is interrupted;
reachedWatermark(5) is below the
+ // watermark(10) already emitted above.
+ watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(20));
+
+ assertThat(emittedElements)
+ .as("watermarks must be non-decreasing")
+ .containsExactly(new Watermark(10));
+ }
+
private static class NoOpInternalTimeServiceManager
implements InternalTimeServiceManager<Object> {
@Override
@@ -106,6 +188,11 @@ class MailboxWatermarkProcessorTest {
return !shouldStopAdvancingFn.test();
}
+ @Override
+ public long getReachedWatermark() {
+ return Long.MIN_VALUE;
+ }
+
@Override
public void snapshotToRawKeyedState(
KeyedStateCheckpointOutputStream stateCheckpointOutputStream,
String operatorName)
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java
index 47bfbe84592..1750c0e20b7 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java
@@ -98,11 +98,15 @@ class UnalignedCheckpointsInterruptibleTimersTest {
assertThat(harness.getOutput())
.containsExactly(
asFiredRecord("key-0"),
+ // Intermediate watermark surfacing progress after
firing the first of
+ // the 2 timers due at firstWindowEnd, before the
drain is interrupted.
+
asWatermark(Instant.ofEpochMilli(firstWindowEnd.toEpochMilli() - 1)),
asMailRecord("key-0"),
asFiredRecord("key-1"),
asMailRecord("key-1"),
asWatermark(firstWindowEnd),
asFiredRecord("key-0"),
+
asWatermark(Instant.ofEpochMilli(secondWindowEnd.toEpochMilli() - 1)),
asMailRecord("key-0"),
asFiredRecord("key-1"),
asMailRecord("key-1"),
@@ -195,6 +199,9 @@ class UnalignedCheckpointsInterruptibleTimersTest {
assertThat(harness.getOutput())
.containsExactly(
asFiredRecord("key-0"),
+ // Intermediate watermark surfacing progress after
firing the first of
+ // the 2 timers due at windowEnd, before the drain
is interrupted.
+
asWatermark(Instant.ofEpochMilli(windowEnd.toEpochMilli() - 1)),
asMailRecord("key-0"),
asFiredRecord("key-1"),
asMailRecord("key-1"),
@@ -203,6 +210,120 @@ class UnalignedCheckpointsInterruptibleTimersTest {
}
}
+ @Test
+ void testIntermediateWatermarksEmittedDuringLongDrain() throws Exception {
+ final Instant t1 = Instant.ofEpochMilli(100L);
+ final Instant t2 = Instant.ofEpochMilli(200L);
+ final Instant t3 = Instant.ofEpochMilli(300L);
+
+ try (final StreamTaskMailboxTestHarness<String> harness =
+ new
StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, Types.STRING)
+ .addJobConfig(
+ CheckpointingOptions.CHECKPOINTING_INTERVAL,
Duration.ofSeconds(1))
+ .addJobConfig(CheckpointingOptions.ENABLE_UNALIGNED,
true)
+ .addJobConfig(
+
CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, true)
+ .modifyStreamConfig(
+
UnalignedCheckpointsInterruptibleTimersTest::setupStreamConfig)
+ .addInput(Types.STRING)
+ .setupOperatorChain(
+ SimpleOperatorFactory.of(
+ new MultipleTimersAtTheSameTimestamp()
+ .withTimers(t1, 1)
+ .withTimers(t2, 1)
+ .withTimers(t3, 1)))
+ .name("first")
+
.finishForSingletonOperatorChain(StringSerializer.INSTANCE)
+ .build()) {
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>("register timers"));
+ harness.processAll();
+ // A single watermark whose drain requires firing multiple,
individually-interrupted
+ // timers (each fired timer schedules a mailbox mail, forcing an
interruption).
+ harness.processElement(asWatermark(t3));
+
+ final List<Watermark> seenWatermarks = new ArrayList<>();
+ while (seenWatermarks.isEmpty()
+ || seenWatermarks.get(seenWatermarks.size() -
1).getTimestamp()
+ < t3.toEpochMilli()) {
+ harness.processSingleStep();
+ Object outputElement;
+ while ((outputElement = harness.getOutput().poll()) != null) {
+ if (outputElement instanceof Watermark) {
+ seenWatermarks.add((Watermark) outputElement);
+ }
+ }
+ }
+
+ // The drain is interrupted after firing each of the 3 timers.
Progress made before
+ // the final interruption should be visible downstream as
intermediate watermarks,
+ // not only as the single final watermark once the whole drain
completes.
+ assertThat(seenWatermarks).hasSizeGreaterThan(1);
+
assertThat(seenWatermarks.get(0).getTimestamp()).isLessThan(t3.toEpochMilli());
+ assertThat(seenWatermarks.get(seenWatermarks.size() -
1).getTimestamp())
+ .isEqualTo(t3.toEpochMilli());
+
assertThat(seenWatermarks).extracting(Watermark::getTimestamp).isSorted();
+ }
+ }
+
+ /**
+ * Once one timer service is interrupted, {@link
+ *
org.apache.flink.streaming.api.operators.InternalTimeServiceManagerImpl#tryAdvanceWatermark}
+ * never even attempts the other services this round, so a
persistently-behind service can
+ * starve the others' contribution to the reported intermediate watermark
for as long as it
+ * itself keeps getting interrupted.
+ */
+ @Test
+ void testStarvedTimerServiceDelaysIntermediateWatermark() throws Exception
{
+ final int timersPerService = 20;
+ final Instant watermark = Instant.ofEpochMilli(timersPerService + 10L);
+
+ try (final StreamTaskMailboxTestHarness<String> harness =
+ new
StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, Types.STRING)
+ .addJobConfig(
+ CheckpointingOptions.CHECKPOINTING_INTERVAL,
Duration.ofSeconds(1))
+ .addJobConfig(CheckpointingOptions.ENABLE_UNALIGNED,
true)
+ .addJobConfig(
+
CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, true)
+ .modifyStreamConfig(
+
UnalignedCheckpointsInterruptibleTimersTest::setupStreamConfig)
+ .addInput(Types.LONG)
+ .setupOperatorChain(
+ SimpleOperatorFactory.of(new
TwoTimerServicesWithEqualBacklogs()))
+ .name("first")
+
.finishForSingletonOperatorChain(StringSerializer.INSTANCE)
+ .build()) {
+ harness.setAutoProcess(false);
+ for (long ts = 1; ts <= timersPerService; ts++) {
+ harness.processElement(new StreamRecord<>(ts));
+ }
+ harness.processAll();
+ harness.processElement(asWatermark(watermark));
+
+ int firedCount = 0;
+ Watermark firstWatermark = null;
+ while (firstWatermark == null && firedCount < 2 *
timersPerService) {
+ harness.processSingleStep();
+ Object outputElement;
+ while ((outputElement = harness.getOutput().poll()) != null) {
+ if (outputElement instanceof Watermark) {
+ firstWatermark = (Watermark) outputElement;
+ break;
+ }
+ firedCount++;
+ }
+ }
+
+ assertThat(firstWatermark).as("a watermark should eventually
appear").isNotNull();
+ // With two equally-backlogged services, an intermediate watermark
should surface well
+ // before either service's entire backlog has drained -- not only
once one of them
+ // (whichever happens to be attempted first) has completely
finished firing.
+ assertThat(firedCount)
+ .as("first watermark should appear before either backlog
is drained")
+ .isLessThan(timersPerService);
+ }
+ }
+
private static Watermark asWatermark(Instant timestamp) {
return new Watermark(timestamp.toEpochMilli());
}
@@ -280,6 +401,56 @@ class UnalignedCheckpointsInterruptibleTimersTest {
}
}
+ /**
+ * Registers one timer per element (at the timestamp given by the
element's value) on each of
+ * two independently-named timer services, each firing one timer at a time
(interrupted via a
+ * scheduled mail).
+ */
+ private static class TwoTimerServicesWithEqualBacklogs extends
AbstractStreamOperator<String>
+ implements OneInputStreamOperator<Long, String>,
+ Triggerable<String, String>,
+ YieldingOperator<String> {
+
+ private transient @Nullable MailboxExecutor mailboxExecutor;
+ private transient InternalTimerService<String> serviceA;
+ private transient InternalTimerService<String> serviceB;
+
+ @Override
+ public boolean useInterruptibleTimers(ReadableConfig config) {
+ return true;
+ }
+
+ @Override
+ public void setMailboxExecutor(MailboxExecutor mailboxExecutor) {
+ super.setMailboxExecutor(mailboxExecutor);
+ this.mailboxExecutor = mailboxExecutor;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ serviceA = getInternalTimerService("serviceA",
StringSerializer.INSTANCE, this);
+ serviceB = getInternalTimerService("serviceB",
StringSerializer.INSTANCE, this);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Long> element) {
+ setCurrentKey("key");
+ long ts = element.getValue();
+ serviceA.registerEventTimeTimer("A", ts);
+ serviceB.registerEventTimeTimer("B", ts);
+ }
+
+ @Override
+ public void onEventTime(InternalTimer<String, String> timer) throws
Exception {
+ mailboxExecutor.execute(() -> {}, "mail");
+ output.collect(asFiredRecord(timer.getNamespace() + "-" +
timer.getTimestamp()));
+ }
+
+ @Override
+ public void onProcessingTime(InternalTimer<String, String> timer)
throws Exception {}
+ }
+
/**
* Registers a timer for the current key on every element; every fired
timer enqueues a mail, so
* that firing is interrupted after each timer.
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
index 19a104a652f..d0e6530c190 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java
@@ -500,6 +500,9 @@ class RowTimeIntervalJoinTest extends
TimeIntervalStreamJoinTestBase {
final List<Object> expectedOutput = new ArrayList<>();
expectedOutput.add(insertRecord(5L, "k1", null, null));
+ // Intermediate watermark surfacing progress after firing the first
timer, before the
+ // drain is interrupted by the test's injected mail.
+ expectedOutput.add(new Watermark(5));
expectedOutput.add(insertRecord(null, null, 6L, "k2"));
expectedOutput.add(insertRecord(7L, "k3", null, null));
expectedOutput.add(insertRecord(null, null, 8L, "k4"));
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/BaseTwoInputStreamOperatorWithStateRetentionTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/BaseTwoInputStreamOperatorWithStateRetentionTest.java
index c0e5f705dc9..0aba8733f1e 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/BaseTwoInputStreamOperatorWithStateRetentionTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/temporal/BaseTwoInputStreamOperatorWithStateRetentionTest.java
@@ -94,6 +94,9 @@ class BaseTwoInputStreamOperatorWithStateRetentionTest {
assertThat(output)
.containsExactly(
firedDesc(0L),
+ // Intermediate watermark surfacing progress after
firing the first of
+ // the 2 timers due at firstWindowEnd, before the
drain is interrupted.
+ "Watermark@" + (firstWindowEnd.toEpochMilli() - 1),
mailDesc(0L),
firedDesc(1L),
mailDesc(1L),
@@ -101,6 +104,7 @@ class BaseTwoInputStreamOperatorWithStateRetentionTest {
firedDesc(0L),
mailDesc(0L),
firedDesc(1L),
+ "Watermark@" + (secondWindowEnd.toEpochMilli() -
1),
mailDesc(1L),
watermarkDesc(secondWindowEnd));
}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
index 15258ff24cb..9e8e3d43e2b 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
@@ -149,15 +149,18 @@ class ProcessSetTableOperatorInterruptibleTimersTest {
recordLabel(3000L, null),
firedLabel(null, 1000L, 5000L),
mailLabel(null, 1000L),
+ firedLabel(NAMED_TIMER, 1000L, 5000L),
+ watermarkLabel(999L),
+ mailLabel(NAMED_TIMER, 1000L),
firedLabel(null, 2000L, 5000L),
mailLabel(null, 2000L),
firedLabel(null, 3000L, 5000L),
+ watermarkLabel(2999L),
mailLabel(null, 3000L),
- firedLabel(NAMED_TIMER, 1000L, 5000L),
- mailLabel(NAMED_TIMER, 1000L),
watermarkLabel(5000L),
recordLabel(6000L, 5000L),
firedLabel(null, 6000L, 7000L),
+ watermarkLabel(5999L),
mailLabel(null, 6000L),
watermarkLabel(7000L));
} else {
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperatorTest.java
index 65eaa89001e..05d29a4d634 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperatorTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperatorTest.java
@@ -83,12 +83,18 @@ class BaseTemporalSortOperatorTest {
assertThat(output)
.containsExactly(
firedDesc(1000L),
+ // Intermediate watermarks surfacing progress
after each timer fires,
+ // before the drain is interrupted by the next
mail.
+ watermarkDesc(999L),
mailDesc(1000L),
firedDesc(2000L),
+ watermarkDesc(1999L),
mailDesc(2000L),
firedDesc(3000L),
+ watermarkDesc(2999L),
mailDesc(3000L),
firedDesc(4000L),
+ watermarkDesc(3999L),
mailDesc(4000L),
watermarkDesc(5000L));
}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/eventtime/EventTimeWatermarkHandlerTest.java
b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/eventtime/EventTimeWatermarkHandlerTest.java
index 36c987701cd..8e866c2d7e5 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/eventtime/EventTimeWatermarkHandlerTest.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/extension/eventtime/EventTimeWatermarkHandlerTest.java
@@ -276,6 +276,11 @@ class EventTimeWatermarkHandlerTest {
throw new UnsupportedOperationException();
}
+ @Override
+ public long getReachedWatermark() {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public void snapshotToRawKeyedState(
KeyedStateCheckpointOutputStream stateCheckpointOutputStream,
String operatorName)
