[
https://issues.apache.org/jira/browse/FLINK-40101?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski closed FLINK-40101.
----------------------------------
Release Note: With unaligned checkpoints and interruptible timers enabled
(execution.checkpointing.unaligned.interruptible-timers.enabled), an operator's
watermark could previously stall for a long time whenever a large backlog of
due timers (e.g. after a rescale) took multiple execution attempts to fully
fire, because progress was only reported once the entire backlog had drained in
one uninterrupted pass. Operators now instead emit an intermediate watermark
reflecting the progress made so far whenever firing that backlog is
interrupted, so downstream operators aren't starved of watermark progress while
it catches up. This behavior is enabled by default. It can be disabled by
setting
execution.checkpointing.unaligned.interruptible-timers.emit-intermediate-watermarks
to false.
Resolution: Fixed
merged commit
[{{695ccca}}|https://github.com/apache/flink/commit/695ccca9a04a11275ffb230024527e3f226b9ab8]
into apache:master
> Watermark forwarding can stall indefinitely under interruptible timers with a
> large due-timer backlog
> -----------------------------------------------------------------------------------------------------
>
> Key: FLINK-40101
> URL: https://issues.apache.org/jira/browse/FLINK-40101
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream, Runtime / Task
> Affects Versions: 2.0.2, 2.3.0, 2.2.1, 1.20.5, 2.1.3
> Reporter: Piotr Nowojski
> Assignee: Piotr Nowojski
> Priority: Major
> Labels: pull-request-available
>
> h3. Summary
> With unaligned checkpoints + interruptible timers enabled (FLINK-20217), an
> operator's output watermark can freeze for a very long time (observed: hours,
> persists across restarts) while input watermark, record processing, and timer
> firing all continue normally and query continue to make progress.
> h3. Symptoms
> - Input watermark advances normally; output watermark frozen at
> Long.MIN_VALUE-ish.
> - Subtask at ~100% CPU actively firing timers (numFiredTimers climbing).
> - Records trickle in at low but genuine rate (not a metric artifact).
> - Started after a scale-up; recurs on every restart.
> h3. Root cause
> InternalTimerServiceImpl#tryAdvanceWatermark only lets
> output.emitWatermark(...) fire once all due timers are drained in one
> uninterrupted pass. Unaligned checkpoint triggers are urgent mailbox mail
> (StreamTask.java:1355-1358) and reliably interrupt that pass once per
> checkpoint interval; ordinary record/watermark ingestion doesn't use the
> mailbox at all, so between checkpoints the loop runs CPU-bound but any single
> interruption discards all watermark-visibility progress for that attempt. A
> large due-timer backlog (e.g. from a rescale-triggered watermark jump) can
> take many checkpoint intervals to drain, during which nothing downstream
> advances — and since the backlog + stuck state are both checkpointed,
> restarts don't help.
> The issue can get even worse and persist as long as there is a backlog of
> records to be processed, even across restarts. When query is restarted,
> subject to watermark strategy, it could happen that first watermark to be
> processed after the restart is much further ahead then the one that was being
> processed before the restart. This prolongs the state where no output
> watermarks are emitted. Thus job can enter a cycle of restarts/rescales,
> where each restart pushes currently fired watermark even more ahead, while
> job never manages to fully process that watermark between restarts.
> In the worst case scenario this can lead to state explosion in the downstream
> operators.
> h3. Note
> The situation is even worse with interruptible timers disabled, as then job
> wouldn't be able to complete checkpoint and it wouldn't be able to make
> progress at all.
> h3. Proposed fix
> Let InternalTimerServiceImpl/InternalTimeServiceManagerImpl report the
> last-fired timestamp as a safe intermediate watermark even when interrupted,
> and have MailboxWatermarkProcessor emit it. Pace this via an internal no-op
> periodic processing-time timer (avoids per-timer currentTimeMillis() and
> fixed-count thresholds); interval configurable, default 5s.
> Affected: flink-runtime (InternalTimerServiceImpl,
> InternalTimeServiceManagerImpl, MailboxWatermarkProcessor); any
> useInterruptibleTimers() operator (TableStreamOperator subclasses,
> CepOperator). Requires
> execution.checkpointing.unaligned.interruptible-timers.enabled (default off).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
