[FLINK-40101][runtime] Emit intermediate watermarks while firing timers#28752
[FLINK-40101][runtime] Emit intermediate watermarks while firing timers#28752pnowojski wants to merge 2 commits into
Conversation
github-actions
Bot
added
the
community-reviewed
| 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); | ||
| } | ||
| } |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Yes, this is doesn't work if there is more than one timer service - which is rare. Only I think Fabian's new operator is using two timer services?
There was a problem hiding this comment.
After giving it the second thought, instead of starting from the same timer service always, I will implement some round robin here. One interruption will only handle one service, without advancing the minReachedWatermark, but every interruption will be alternating between different timer services and subsequently advancing the minReachedWatermark across multiple calls. Given that the interruptions will happen every ~200ms (default auto watermarking interval), that should be perfectly fine. WDYT?
There was a problem hiding this comment.
Yes, this is doesn't work if there is more than one timer service - which is rare. Only I think Fabian's new operator is using two timer services?
AbstractProcessTableOperator for PTFs?
There was a problem hiding this comment.
Oh, and can we stop the iteration ince fullyAdvanced is false?
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.
rkhachatryan
left a comment
rkhachatryan
left a comment
There was a problem hiding this comment.
Thanks for updating the PR, LGTM.
| 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); | ||
| } | ||
| } |
There was a problem hiding this comment.
Yes, this is doesn't work if there is more than one timer service - which is rare. Only I think Fabian's new operator is using two timer services?
AbstractProcessTableOperator for PTFs?
4 participants
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:
@Public(Evolving): (yes / no)Documentation
Was generative AI tooling used to co-author this PR?
Yes, claude 5.0