[
https://issues.apache.org/jira/browse/FLINK-39521?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rui Fan updated FLINK-39521:
----------------------------
Description:
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.
was:Introduces `OutputWriter` / `OutputWriterImpl`, which routes post-filter
bytes into one of three paths per `write()`: fill the active Network Buffer
(P1), spill to disk when no buffer is available (P2), or eagerly drain disk
back into a buffer when one frees up (P3). `flush()` seals the active buffer;
`close()` performs a blocking drain to exhaust remaining disk data;
`checkpointPendingEntries()` streams all queued disk entries into a checkpoint
in one sequential pass.
> 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
>
> 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)
