[jira] [Closed] (FLINK-40101) Watermark forwarding can stall indefinitely under interruptible timers with a large due-timer backlog

["Piotr Nowojski (Jira)" <ji...@apache.org>]

     [ 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)