[
https://issues.apache.org/jira/browse/FLINK-41073?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130438#comment-18130438
]
Shrey Bansal edited comment on FLINK-41073 at 10/22/26, 5:07 PM:
-----------------------------------------------------------------
I’d like to work on this. Here’s my understanding of the problem and a proposed
approach.
*What happens today*
In {{{}CheckpointCoordinator#receiveAcknowledgeMessage{}}}, the subtask state’s
shared handles are registered before {{PendingCheckpoint#acknowledgeTask}}
records the ack. If {{registerSharedStates}} throws, for example
{{SharedStateRegistryImpl#registerReference}} with “StateObjects underlying
same key should be equal !” as in FLINK-38574, the exception leaves the method
before the following recording of ack.
{{ExecutionGraphHandler#processCheckpointCoordinatorMessage}} runs this on the
io executor and only logs the exception, so the pending checkpoint keeps
waiting for that task until the {{CheckpointCanceller}} expires it at the
checkpoint timeout.
*Proposal*
* Catch the exception around {{registerSharedStates}} in
{{{}receiveAcknowledgeMessage{}}}. We already hold the coordinator lock there.
* If the checkpoint is still pending, abort it right away via
{{{}abortPendingCheckpoint{}}}, the same path a decline takes. The timeout is
cancelled, the failure is reported to the {{{}CheckpointFailureManager{}}}, the
tasks are notified, and the next checkpoint can be triggered.
* Use a new {{CheckpointFailureReason}} for it, and count it against the
tolerable failures, since today the same situation already ends as a counted
{{{}CHECKPOINT_EXPIRED{}}}, only later.
* Discard the failing acknowledgement’s state with
{{{}discardSubtaskState{}}}, like the existing UNKNOWN and DISCARDED branches.
*Open question*
New shared files after the failing handle are neither registered nor discarded,
so they can leak. Deleting them would have to skip placeholders, which point at
files that earlier checkpoints still use, so I’d lean towards leaving them, as
{{AsyncCheckpointRunnable#declineWithoutDiscard}} does for an unconfirmed ack.
Would you prefer handling them in this change?
*Testing*
A test in {{CheckpointCoordinatorTest}} where registration throws, asserting
that the pending checkpoint is aborted immediately with the new reason (not on
timeout), and that the acknowledgement’s private state is discarded. I’ll check
that it fails on master first.
[~martijnvisser], does this direction sound right? If so, could you assign the
ticket to me?
was (Author: JIRAUSER315107):
I’d like to work on this. Here’s my understanding of the problem and a proposed
approach.
*What happens today*
In {{{}CheckpointCoordinator#receiveAcknowledgeMessage{}}}, the subtask state’s
shared handles are registered before {{PendingCheckpoint#acknowledgeTask}}
records the ack. If {{registerSharedStates}} throws, for example
{{SharedStateRegistryImpl#registerReference}} with “StateObjects underlying
same key should be equal !” as in FLINK-38574, the exception leaves the method
before the following recording of ack.
{{ExecutionGraphHandler#processCheckpointCoordinatorMessage}} runs this on the
io executor and only logs the exception, so the pending checkpoint keeps
waiting for that task until the {{CheckpointCanceller}} expires it at the
checkpoint timeout.
*Proposal*
* Catch the exception around {{registerSharedStates}} in
{{{}receiveAcknowledgeMessage{}}}. We already hold the coordinator lock there.
* If the checkpoint is still pending, abort it right away via
{{{}abortPendingCheckpoint{}}}, the same path a decline takes. The timeout is
cancelled, the failure is reported to the {{{}CheckpointFailureManager{}}}, the
tasks are notified, and the next checkpoint can be triggered.
* Use a new {{CheckpointFailureReason}} for it, and count it against the
tolerable failures, since today the same situation already ends as a counted
{{{}CHECKPOINT_EXPIRED{}}}, only later.
* Discard the failing acknowledgement’s state with
{{{}discardSubtaskState{}}}, like the existing UNKNOWN and DISCARDED branches.
Because the handle marks itself registered before the loop, this removes its
private and meta state but leaves shared files to the
{{{}SharedStateRegistry{}}}. Handles registered before the failure are removed
on subsumption, as for any registered handle.
*Open question*
New shared files after the failing handle are neither registered nor discarded,
so they can leak. Deleting them would have to skip placeholders, which point at
files that earlier checkpoints still use, so I’d lean towards leaving them, as
{{AsyncCheckpointRunnable#declineWithoutDiscard}} does for an unconfirmed ack.
Would you prefer handling them in this change?
*Testing*
A test in {{CheckpointCoordinatorTest}} where registration throws, asserting
that the pending checkpoint is aborted immediately with the new reason (not on
timeout), and that the acknowledgement’s private state is discarded. I’ll check
that it fails on master first.
[~martijnvisser], does this direction sound right? If so, could you assign the
ticket to me?
> Checkpoint acknowledgement that fails shared state registration is dropped
> and the checkpoint waits for its timeout
> -------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41073
> URL: https://issues.apache.org/jira/browse/FLINK-41073
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing
> Affects Versions: 2.4.0
> Reporter: Martijn Visser
> Priority: Major
>
> In CheckpointCoordinator#receiveAcknowledgeMessage the subtask state
> registers its shared handles before the task is acknowledged. When
> registerSharedStates throws, as it did in FLINK-38574,
> ExecutionGraphHandler#processCheckpointCoordinatorMessage only logs the
> exception and the acknowledgement is lost. The pending checkpoint then waits
> for its full timeout (ten minutes by default) before it fails, the handles
> registered up to the failing one stay registered under a checkpoint that
> never completes, and the handles after it are neither registered nor
> discarded. FLINK-40663 shows what that looks like, a 600 second hang per
> occurrence.
> We could consider failing the pending checkpoint right away with its own
> CheckpointFailureReason and discard the state of that acknowledgement, so the
> job sees the failure within seconds and nothing leaks.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
