Piotr Nowojski created FLINK-40101:
--------------------------------------
Summary: 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.1.3, 1.20.5, 2.2.1, 2.3.0, 2.0.2
Reporter: Piotr Nowojski
Assignee: Piotr Nowojski
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)
