[
https://issues.apache.org/jira/browse/FLINK-40101?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40101:
-----------------------------------
Labels: pull-request-available (was: )
> 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)
