[
https://issues.apache.org/jira/browse/FLINK-39521?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-39521:
-----------------------------------
Labels: pull-request-available (was: )
> Support push-based recovery in input channels
> ---------------------------------------------
>
> Key: FLINK-39521
> URL: https://issues.apache.org/jira/browse/FLINK-39521
> Project: Flink
> Issue Type: Sub-task
> Reporter: Rui Fan
> Assignee: Rui Fan
> Priority: Major
> Labels: pull-request-available
>
> Teach the physical Local/RemoteInputChannel to be created directly in a
> recovery state and receive recovered buffers via a new
> RecoverableInputChannel interface; recovered-channel conversion hands buffers
> over through that interface instead of migrating them via constructors.
> Scope:
> * New RecoverableInputChannel contract (push append,
> finishRecoveredBufferDelivery, insertRecoveryCheckpointBarrierIfInRecovery,
> requestRecoveryBufferBlocking, onRecoveredStateConsumed);
> getStateConsumedFuture unified on the InputChannel base class.
> * Two new recovery sentinels with dedicated EventSerializer tags:
> RecoveryCheckpointBarrier (per-checkpoint marker inside the recovery queue)
> and EndOfFetchedChannelStateEvent (tail sentinel).
> * LocalInputChannel / RemoteInputChannel: full in-recovery state machine —
> recovered-buffer queueing, live-data hiding, upstream-readiness gating,
> credit suppression with event stash (remote), checkpointStarted split into
> in-recovery / normal branches with collectPreRecoveryBarrier.
> * Gates: requestPartitions(needsRecovery), conversion via the push
> interface, state-consumed future aggregation.
> * BufferManager: credit notification gated behind notifyInitiallyEnabled.
> * Netty: PartitionRequest carries needsRecovery (wire-format change, +1 byte
> per message — safe under the single-version-cluster assumption of the TM
> network protocol); the producer-side view reader starts with zero credit
> while the consumer's exclusive buffers are lent to recovery.
> All new paths are gated on needsRecovery/inRecovery and nothing passes
> needsRecovery=true, so default behavior is unchanged. The v1
> conversion-migration path is retired here; the (experimental, default-false)
> execution.checkpointing.during-recovery.enabled flag is therefore temporarily
> removed from ITCase randomization while the flag-on path is being rewired.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
