[
https://issues.apache.org/jira/browse/FLINK-39522?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rui Fan updated FLINK-39522:
----------------------------
Description:
Restructure StreamTask channel-state recovery into an asynchronous future chain
on the channelIOExecutor and unify recovery completion on the gates'
state-consumed futures.
Behavior change (checkpointing-during-recovery enabled only): the task now
consumes recovered state only after rescale filtering has fully completed —
previously consumption overlapped with filtering. Filtered buffers are held in
the recovered channels' memory queues and handed to the physical channels
through the push interface at conversion. Rationale: a checkpoint taken during
recovery can only proceed once filtering is complete; making the order strict
greatly simplifies the coordination between filtering and checkpointing.
Scope:
* Split restoreStateAndGates into recoverChannelsWithCheckpointing /
recoverChannelsWithoutCheckpointing; recovery completion suspends the restore
mailbox loop; recovery failures are rethrown from restoreInternal instead of
being swallowed.
* Flag-off path preserves master behavior (per-gate conversion off each gate's
own state-consumed future — also avoids a selective-reading multi-input
deadlock).
* Remove the now-unused filtering-progress gate API
(getBufferFilteringCompleteFuture, set/isCheckpointingDuringRecoveryEnabled).
* Defer task finish until recovery completes (fixes an existing race where a
finite task could finish before recovery completed; also affects the non-CDR
path as a hardening).
was:`RecoveredInputChannel`, `LocalInputChannel`, and `RemoteInputChannel`
now read recovery buffers via `store.tryTake()` instead of maintaining their
own recovered-buffer queue. Removes the `onRecoveredStateBuffer()` callback and
the `initialRecoveredBuffers` hand-off on channel conversion, wires
`store.checkpoint()` into `checkpointStarted()`, and routes in-use counts /
`releaseAllResources()` through the store.
> Rework StreamTask channel-state recovery into an async future chain; start
> consumption only after filtering completes
> ---------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-39522
> URL: https://issues.apache.org/jira/browse/FLINK-39522
> Project: Flink
> Issue Type: Sub-task
> Reporter: Rui Fan
> Assignee: Rui Fan
> Priority: Major
>
> Restructure StreamTask channel-state recovery into an asynchronous future
> chain on the channelIOExecutor and unify recovery completion on the gates'
> state-consumed futures.
> Behavior change (checkpointing-during-recovery enabled only): the task now
> consumes recovered state only after rescale filtering has fully completed —
> previously consumption overlapped with filtering. Filtered buffers are held
> in the recovered channels' memory queues and handed to the physical channels
> through the push interface at conversion. Rationale: a checkpoint taken
> during recovery can only proceed once filtering is complete; making the order
> strict greatly simplifies the coordination between filtering and
> checkpointing.
> Scope:
> * Split restoreStateAndGates into recoverChannelsWithCheckpointing /
> recoverChannelsWithoutCheckpointing; recovery completion suspends the restore
> mailbox loop; recovery failures are rethrown from restoreInternal instead of
> being swallowed.
> * Flag-off path preserves master behavior (per-gate conversion off each
> gate's own state-consumed future — also avoids a selective-reading
> multi-input deadlock).
> * Remove the now-unused filtering-progress gate API
> (getBufferFilteringCompleteFuture, set/isCheckpointingDuringRecoveryEnabled).
> * Defer task finish until recovery completes (fixes an existing race where a
> finite task could finish before recovery completed; also affects the non-CDR
> path as a hardening).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
