[
https://issues.apache.org/jira/browse/FLINK-41299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136365#comment-18136365
]
Chen Zhang commented on FLINK-41299:
------------------------------------
Why FLINK-37670 does not cover this (and appears to have exposed it):
* Before 1.20.2, {{SourceCoordinator#announceCombinedWatermark}} skipped
subtasks with {{context.hasNoMoreSplits(subtaskId)}}. That check means "the
coordinator has nothing left to assign", not "the subtask is done", which
caused the FLINK-37670 deadlock (readers still processing assigned splits
stopped getting {{maxAllowedWatermark}} updates). As a side effect, though,
finished bounded readers never received alignment events.
* FLINK-37670 removed the skip and now sends to every subtask in the
aggregator, relying on sends to non-running tasks failing harmlessly with
{{TaskNotRunningException}}. Comparing the 1.19.3 and 1.20.5 bytecode confirms
the {{hasNoMoreSplits}} call is gone.
* That assumption only holds if the event is actually sent. A checkpoint calls
{{markForCheckpoint}} / {{tryCloseGateway}} on every subtask gateway, FINISHED
ones included, and a gateway reopens only when its subtask reports barrier
injection ({{handleEventFromOperator}}), which a FINISHED subtask never does.
Later alignment events for that subtask are therefore queued in
{{blockedEventsMap}} and never sent. No {{TaskNotRunningException}} is ever
raised; the futures just stay in {{unconfirmedEvents}}, and
{{completeCheckpointOnceEventsAreDone}} waits on them forever.
* Net effect: before 1.20.2 the {{hasNoMoreSplits}} skip hid this issue while
causing FLINK-37670; after 1.20.2 it is the reverse.
Suggested direction that satisfies both: decide by execution state rather than
split assignment. Keep sending to every RUNNING subtask (preserving the
FLINK-37670 fix), and ensure events to FINISHED subtasks cannot be parked
behind a closed gateway. Concretely: skip them in
{{announceCombinedWatermark}}, and/or have {{OperatorCoordinatorHolder}} not
mark/close FINISHED subtasks' gateways at checkpoint time and not count their
events as unconfirmed. The second part also covers the subtask-is-finishing
race mentioned in FLINK-37670.
> 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: 2.1.0, 2.0.1, 2.2.0, 2.3.0, 1.20.5
> Reporter: Chen Zhang
> Priority: Major
>
> 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)
