[
https://issues.apache.org/jira/browse/FLINK-40101?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski updated FLINK-40101:
-----------------------------------
Description:
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).
was:
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).
> 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
>
> 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)
