Chen Zhang created FLINK-41299:
----------------------------------
Summary: Watermark alignment deadlocks checkpoints after source
subtasks finish
Key: FLINK-41299
URL: https://issues.apache.org/jira/browse/FLINK-41299
Project: Flink
Issue Type: Bug
Affects Versions: 1.20.5, 2.3.0, 2.2.0, 2.0.1, 2.1.0
Reporter: Chen Zhang
h3. Summary
With watermark alignment enabled on a FLIP-27 source and
{{execution.checkpointing.checkpoints-after-tasks-finish}} enabled (the
default), a bounded job whose source subtasks reach end-of-input at different
times may never finish. After some subtasks are FINISHED and one more
checkpoint (or savepoint) has been taken, every subsequent checkpoint gets
stuck in the operator-coordinator phase:
{noformat}
OperatorCoordinatorHolder - Coordinator checkpoint <N> for coordinator <id> is
awaiting <M> pending events
{noformat}
The checkpoint is never triggered on the tasks, so source subtasks that reach
end-of-input later stay forever in {{StreamTask#afterInvoke}} waiting for their
final checkpoint (FLIP-147). The job neither finishes nor fails; checkpoints
time out and are retried indefinitely.
h3. Environment
* Flink 1.20.5 (presumably every version containing FLINK-37670, i.e. 1.20.2+,
2.0.1+, 2.1.0+)
* Bounded FLIP-27 source (observed with a {{HybridSource}} whose last
underlying source is a bounded split-based source)
* {{WatermarkStrategy#withWatermarkAlignment(group, maxDrift, updateInterval)}}
* {{execution.checkpointing.checkpoints-after-tasks-finish: true}} (default),
periodic checkpoints (and optionally periodic savepoints)
* Uneven work across subtasks, so some subtasks reach end-of-input well before
others
* Reproduces with and without HA
h3. What happens
# A group of subtasks reaches end-of-input; the next checkpoint completes and
they transition to FINISHED.
# {{SourceCoordinator#announceCombinedWatermark}} keeps sending a
{{WatermarkAlignmentEvent}} on every update interval to all subtasks in the
watermark aggregator, including the FINISHED ones. This happens even when the
announced value is {{Long.MAX_VALUE}}, i.e. when alignment is not throttling
anything.
# The next checkpoint (or savepoint) calls
{{SubtaskGatewayImpl#markForCheckpoint}} on every subtask gateway and, after
the coordinator snapshot, {{tryCloseGateway}}. A gateway is re-opened only in
{{OperatorCoordinatorHolder#handleEventFromOperator}} (when that subtask
reports barrier injection) or on abort/reset. A FINISHED subtask never injects
a barrier, so its gateway stays closed permanently.
# From then on, every alignment event addressed to a FINISHED subtask is parked
in that gateway's {{blockedEventsMap}}, and its future is tracked in
{{OperatorCoordinatorHolder#unconfirmedEvents}}. It is never sent, so it never
completes. (An event that is actually sent to a FINISHED execution fails fast
with {{TaskNotRunningException}} and is dropped from the tracker; the problem
is specifically events buffered behind a closed gateway.)
# On the next checkpoint, {{completeCheckpointOnceEventsAreDone}} finds M
incomplete futures, logs "awaiting M pending events" and waits on
{{FutureUtils#waitForAll}} indefinitely. The {{CheckpointCoordinator}} never
triggers the tasks.
# The remaining subtasks have already processed all their data ({{endInput}}
called, sink flushed) and wait in {{StreamTask#afterInvoke}} →
{{MailboxProcessor#runMailboxLoop}} for {{finalCheckpointCompleted}}, which
never comes. (Their busy-time metric reports ~1000 ms/s because waiting in
{{afterInvoke}} is not counted as idle time.)
The pending count matches exactly {{#FINISHED subtasks × (time since the
gateway-closing checkpoint / alignment update interval)}}. Observed in three
consecutive runs (parallelism 20, update interval 5 s):
||Run||FINISHED subtasks||Time since gateway-closing checkpoint||Pending events
logged||Expected||
|1|13|~495 s|1287|13 × 99 = 1287|
|2|13|~330 s|858|13 × 66 = 858|
|3|14|~215 s|602|14 × 43 = 602|
In these runs the gateway-closing checkpoint happened to be a periodic
savepoint, which additionally does not count as a final checkpoint for subtasks
that were already at end-of-input. The same happens with a regular checkpoint
taken while some subtasks are still reading. Restarting the job reproduces the
hang deterministically, because the restored split assignment is the same.
h3. Why this looks like a regression
In 1.19.x, {{SourceCoordinator#announceCombinedWatermark}} skipped subtasks for
which {{SourceCoordinatorContext#hasNoMoreSplits(subtaskId)}} returned true, so
finished bounded readers did not receive alignment events and nothing
accumulated behind their gateways. FLINK-37670 removed that check (it caused a
different alignment deadlock), assuming that sending to non-running tasks would
fail harmlessly. That assumption does not hold once a checkpoint has closed the
finished subtask's gateway: the events are never sent, never fail, and stay
unconfirmed. Comparing {{SourceCoordinator#announceCombinedWatermark}} between
1.19.3 and 1.20.5 shows the {{hasNoMoreSplits}} call is gone in 1.20.5, while
the gateway handling in {{OperatorCoordinatorHolder}} / {{SubtaskGatewayImpl}}
is unchanged.
h3. Expected behavior
Source subtasks that finish at different times under watermark alignment should
all be able to finish; events addressed to FINISHED subtasks must not block
coordinator checkpoints.
h3. Possible fixes
* In {{SourceCoordinator#announceCombinedWatermark}}, do not send alignment
events to subtasks whose execution is FINISHED (based on execution state /
subtask readiness rather than {{hasNoMoreSplits}}), and/or remove finished
subtasks from the watermark aggregator (related: FLINK-35157, FLINK-38477,
FLINK-40505).
* In {{OperatorCoordinatorHolder}}: do not mark/close the gateways of FINISHED
subtasks on checkpoint, or do not track events to FINISHED subtasks in
{{unconfirmedEvents}} (complete or drop them, as already happens when sending
to a non-running execution).
h3. Workarounds
* {{execution.checkpointing.checkpoints-after-tasks-finish: false}}: finished
tasks exit without a final checkpoint, so the stuck coordinator checkpoint no
longer blocks job completion (verified: the job finished within minutes). This
loses the final-checkpoint guarantee for two-phase-commit sinks.
* Disable watermark alignment for bounded jobs.
* Have all readers emit end-of-input together (hold END_OF_INPUT until the
enumerator has seen every reader finish).
h3. Relevant logs
JobManager:
{noformat}
CheckpointCoordinator - Triggering checkpoint 42
(type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD})
@ <ts> for job <job-id>.
OperatorCoordinatorHolder - Coordinator checkpoint 42 for coordinator <id> is
awaiting 1287 pending events
SourceCoordinator - Distributing maxAllowedWatermark=9223372036854775807 of
group=<group> to subTaskIds=[0, 1, 2, ..., 19] for source <source>.
SourceCoordinator - Distributing maxAllowedWatermark=9223372036854775807 ...
(every update interval, all 20 subtasks, 13 of them FINISHED)
{noformat}
Thread dump of a remaining (not yet FINISHED) source task:
{noformat}
"Source: <source> -> ... (9/20)#0" WAITING
at jdk.internal.misc.Unsafe.park(Native Method)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1764)
at
org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl.take(TaskMailboxImpl.java:149)
at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:379)
at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:368)
at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:1003)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:923)
{noformat}
On the task side, these subtasks had already received {{endInput}}, and their
source readers never logged a snapshot for the stuck checkpoint (they were
never triggered).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
