[FLINK-40101][runtime] Emit intermediate watermarks while firing timers by pnowojski · Pull Request #28752 · apache/flink

pnowojski · 2026-07-15T16:19:58Z

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.

Verifying this change

Added new unit tests and is covered by various existing unit/IT tests.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

Was generative AI tooling used to co-author this PR?

Yes, claude 5.0

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.

flinkbot · 2026-07-15T16:26:56Z

CI report:

3782eea Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Efrat19 · 2026-07-15T17:57:10Z

+                    ConfigOptions.key(
+                                    "execution.checkpointing.unaligned.interruptible-timers.intermediate-watermark-interval")
+                            .durationType()
+                            .defaultValue(Duration.ofSeconds(5))


Isn't 5s a long wait for WM progress?
being X25 pipeline.auto-watermark-interval default

🤔 Maybe... maybe this should be changed to a boolean flag, and the interval should taken directly from pipeline.auto-watermark-interval?

I've picked 5s default here, because anything > 1s doesn't matter for performance. 200ms would have marginal effect if any, hard to say without benchmarking. However indeed it seems as if re-using here pipeline.auto-watermark-interval makes sense, as it has almsot the same performance implications vs trying to keep up with real time?

Previously I've thought this features matter only if there is a huge backpressure, and huge watermark stall. However if we have a lot of fine grained timers to fire, smaller intervals here can smooth out workload for downstream operators. For example if job is generally speaking keeping up with the pending records, but there is an occasional couple of seconds hiccup when a lot of timers fire.

I've implemented the change to use auto-watermark-interval - please check fixup commit.

I'd keep the .intermediate-watermark-interval and only make it equal to pipeline.auto-watermark-interval by default
But either way works

I think reducing the amount of knobs is a good idea here.

Future us would appreciate this

…ng timers

rkhachatryan · 2026-07-16T17:51:58Z

+                    ConfigOptions.key(
+                                    "execution.checkpointing.unaligned.interruptible-timers.emit-intermediate-watermarks")
+                            .booleanType()
+                            .defaultValue(true)


I think this should be disabled by default for 1 release to minimize potential impact.

rkhachatryan · 2026-07-16T17:57:59Z

        for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) {
-            if (!service.tryAdvanceWatermark(watermark.getTimestamp(), shouldStopAdvancingFn)) {
-                return false;
+            // 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: a service we don't retry this round may be even further behind.
+            if (fullyAdvanced) {
+                fullyAdvanced =
+                        service.tryAdvanceWatermark(
+                                watermark.getTimestamp(), shouldStopAdvancingFn);
            }
        }


Sorry but I don't understand this loop :)

First, once fullyAdvance is false, we still iterate, but just do nothing? Why not just break?

Second, if the iteration order is the same next round (likely) then we won't update min wm if there are more services with lower wm after the interrupted one.
I think in practice this renders the feature as no-op with high probability.

rkhachatryan · 2026-07-16T18:00:30Z

+            if (getContainingTask()
+                    .getJobConfiguration()
+                    .get(
+                            CheckpointingOptions
+                                    .UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS)) {
+                timeServiceManager.configureIntermediateWatermarkInterval(
+                        Duration.ofMillis(getExecutionConfig().getAutoWatermarkInterval()));
+            }


What about V2?

rkhachatryan · 2026-07-16T18:01:18Z

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


Should we also update lastEmittedIntermediateWatermark in the short-cut branch above?

rkhachatryan · 2026-07-16T18:02:03Z

+            // 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));


Can we honor the config option here?

Efrat19 reviewed Jul 15, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Jul 16, 2026

fixup! [FLINK-40101][runtime] Emit intermediate watermarks while firi…

3782eea

…ng timers

pnowojski force-pushed the f40101 branch from f5b7fbb to 3782eea Compare July 16, 2026 15:24

rkhachatryan reviewed Jul 16, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-40101][runtime] Emit intermediate watermarks while firing timers#28752

[FLINK-40101][runtime] Emit intermediate watermarks while firing timers#28752
pnowojski wants to merge 2 commits into
apache:masterfrom
pnowojski:f40101

pnowojski commented Jul 15, 2026

Uh oh!

flinkbot commented Jul 15, 2026 •

edited

Loading

Uh oh!

Efrat19 Jul 15, 2026

Uh oh!

pnowojski Jul 16, 2026

Uh oh!

pnowojski Jul 16, 2026

Uh oh!

Efrat19 Jul 16, 2026

Uh oh!

pnowojski Jul 16, 2026

Uh oh!

Efrat19 Jul 16, 2026

Uh oh!

rkhachatryan Jul 16, 2026

Uh oh!

rkhachatryan Jul 16, 2026 •

edited

Loading

Uh oh!

rkhachatryan Jul 16, 2026

Uh oh!

rkhachatryan Jul 16, 2026

Uh oh!

rkhachatryan Jul 16, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

pnowojski commented Jul 15, 2026

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rkhachatryan Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Jul 15, 2026 •

edited

Loading

rkhachatryan Jul 16, 2026 •

edited

Loading