[FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes#28659
Open
1996fanrui wants to merge 16 commits into
Open
[FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes#286591996fanrui wants to merge 16 commits into
1996fanrui wants to merge 16 commits into
Conversation
…nputs finished StreamMultipleInputProcessor.processInput returned NOTHING_AVAILABLE when no input was selectable, even once all inputs were finished. Since getAvailableFuture() returns AVAILABLE as soon as all inputs are finished, the mailbox loop then never blocks and would spin on processInput at 100% CPU if it were ever invoked again after the operator finished. This is a latent inconsistency with single-input processors (which keep returning END_OF_INPUT once finished) and is independent of any particular feature; it is just not reached today because a task suspends on the first END_OF_INPUT and processInput is not called again. The checkpointing-during- recovery work (later in this series) does re-invoke processInput after an operator finishes during recovery, which would otherwise turn this into a hang. Report END_OF_INPUT when there is nothing to read and all inputs are finished, consistent with single-input processors and MultipleInputSelectionHandler#calculateOverallStatus.
…m toBeConsumedBuffers
Split the single toBeConsumedBuffers queue into two queues with disjoint
responsibilities:
- recoveredBuffers (new): holds buffers migrated from RecoveredInputChannel
during construction; consumed by getNextRecoveredBuffer() which retains
the priority-event interleaving and last-buffer dynamic next-data-type
detection introduced by FLINK-39018.
- toBeConsumedBuffers (existing): reverted to its pre-FLINK-39018 role of
holding FullyFilledBuffer partial-buffer splits only. The recovery-aware
early branch in getNextBuffer() and the checkpointStarted inflight scan
no longer touch this queue.
Restores the checkState(toBeConsumedBuffers.isEmpty()) guard in
requestSubpartitions() (removed by cebc174). hasPendingPriorityEvent,
notifyPriorityEvent, and the constructor signature are unchanged.
Pure refactor: no public API change, no new tests; verified by the 9 existing
LocalInputChannelTest regression cases.
…andler with concrete no-filtering/filtering handlers Single-file class split of RecoveredChannelStateHandler.java; all classes stay package-private in this file. No behavior change. - New abstract AbstractInputChannelRecoveredStateHandler: fields inputGates, channelMapping, rescaledChannels, oldToNewMappings; methods getMappedChannels, calculateMapping, getChannel - verbatim extraction from InputChannelRecoveredStateHandler. Adds a closeInternal() template hook used by later phases. - New NoSpillingHandler: recover() is the verbatim old non-filtering branch (onRecoveredStateBuffer(descriptor) followed by onRecoveredStateBuffer(buffer.retainBuffer()), same try/finally recycle); inherits the verbatim network-pool getBuffer. - New FilteringHandler (the v1 filtering behavior, verbatim): recover() = old recoverWithFiltering (filteringHandler.filterAndRewrite(..., channel::requestBufferBlocking) delivering List<Buffer> via onRecoveredStateBuffer); getBuffer = old getPreFilterBuffer (reusable heap MemorySegment, preFilterBufferInUse invariant); closeInternal = old segment-freeing close. - New static create(...) factory selecting NoSpillingHandler vs FilteringHandler - equivalent to the old internal if-branch. SequentialChannelStateReaderImpl switches from direct construction to the factory (mechanical). - ResultSubpartitionRecoveredStateHandler: untouched. - Trace label in RecoveredInputChannel.onRecoveredStateBuffer: "InputChannelRecoveredStateHandler#recover" -> "NoSpillingHandler#recover" (final label). - Tests: mechanical adaptation of InputChannelRecoveredStateHandlerTest, RecoveredChannelStateHandlerTest.
- InputGate: new abstract InputChannel getChannel(InputChannelInfo). - SingleInputGate: channels[channelInfo.getInputChannelIdx()]. UnionInputGate: resolve via inputGatesByGateIndex.get(channelInfo.getGateIdx()) .getChannel(channelInfo) (correct global-vs-member index semantics). InputGateWithMetrics: delegate. CheckpointedInputGate: delegate helper. - AbstractStreamTaskNetworkInput: getChannel(channelInfo.getInputChannelIdx()) -> getChannel(channelInfo). - Test impls: MockInputGate, MockIndexedInputGate, AlignedCheckpointsMassiveRandomTest's inner gate.
…ources visibility - NetworkActionsLogger: additive tracePersist(String, Object, Object, long) overload; the existing Buffer overload delegates to it. - RecoveredInputChannel#releaseAllResources: package-private -> public.
…se randomization Drop the randomize(conf, CHECKPOINTING_DURING_RECOVERY_ENABLED, true, false) line from TestStreamEnvironment.randomizeConfiguration (the flag stays at its default false unless a test sets it explicitly) and @disabled the explicitly flag-on RecoveredStateFilteringLargeRecordITCase. Rationale: a following commit of this PR retires the v1 conversion-migration path, so ~50% of randomized ITCases would otherwise run flag-on and fail until the StreamTask recovery rework and the checkpoint-coordination-during-recovery PRs land. The randomization and the ITCase are restored in the coordination PR (PR4 of the series).
… notifyInitiallyEnabled Add a notifyInitiallyEnabled constructor parameter to BufferManager and a new enableNotify() method: while notification is disabled, requested and recycled floating buffers are queued but never announced to the producer as credit; enableNotify() opens the gate and announces the queued buffers atomically with respect to concurrent recycle/floating-buffer callbacks. All existing callers (RecoveredInputChannel, RemoteInputChannel) pass true, so behavior is unchanged. The disabled mode is used by a later commit of this PR, where a remote channel created in recovery state withholds credit until its recovered state has been consumed.
…ery sentinels Introduce the push-based recovery contract and its sentinels: - RecoverableInputChannel (@internal): getChannelInfo(), onRecoveredStateBuffer(Buffer), finishRecoveredBufferDelivery(), insertRecoveryCheckpointBarrierIfInRecovery(long), requestRecoveryBufferBlocking(), onRecoveredStateConsumed(). - EndOfFetchedChannelStateEvent: singleton RuntimeEvent tail sentinel of the recovered-buffer stream; reflective write()/read() throw; deliberately distinct from EndOfInputChannelStateEvent. - RecoveryCheckpointBarrier (checkpoint.channel): per-checkpoint sentinel carried inside the recovery queue; write()/read() throw. - EventSerializer: type tags 13 (RECOVERY_CHECKPOINT_BARRIER_EVENT) and 14 (END_OF_FETCHED_CHANNEL_STATE_EVENT). Recovery completion is unified on the InputChannel base class: the new abstract getStateConsumedFuture() completes once a channel has consumed all of its recovered state (or never had any / was released). It is abstract rather than defaulted so a future subclass must state its recovery semantics explicitly. RecoveredInputChannel's package-private getter becomes a public override (field tightened to CompletableFuture<Void>); UnknownInputChannel returns a completed future (a channel with persisted state is always a RecoveredInputChannel at restore time). Local/RemoteInputChannel temporarily return a completed future until their push-based recovery state lands in the next commits. TestInputChannel implements the new contract as a spying test double.
Teach LocalInputChannel to be created directly in a recovery state and to receive recovered buffers through the RecoverableInputChannel interface, instead of migrating them via the constructor: - Constructor: drop ArrayDeque<Buffer> initialRecoveredBuffers, add networkBuffersPerChannel and needsRecovery. With needsRecovery=false the stateConsumedFuture is pre-completed, inRecovery stays false and no recovery BufferManager exists -- behavior is byte-for-byte unchanged. - recoveredBuffers becomes a Deque<Buffer> whose monitor also guards inRecovery and recoverySequenceNumber (starts at Integer.MIN_VALUE). - RecoverableInputChannel implementation: push append with released-state recycling; finishRecoveredBufferDelivery() waits for upstreamReady and appends the EndOfFetchedChannelStateEvent sentinel; insertRecoveryCheckpointBarrierIfInRecovery(); requestRecoveryBufferBlocking() lends exclusive buffers via the recovery-only BufferManager (setup() requests them) -- no production caller until the drainer PR; onRecoveredStateConsumed() flips recovery off and completes the stateConsumedFuture. - getNextBuffer() is rewritten under a single inRecovery predicate: recovered buffer first, then a pending priority event pulled from the subpartition view, otherwise empty (live data stays hidden while in recovery). wrapRecoveredBufferAsAvailability() materializes FileRegionBuffer/CompositeBuffer, assigns recovery sequence numbers and computes the next data type via peekNextDataType. - upstreamReady completes on subpartition-view creation; completeUpstreamReadyForTest() for tests. - checkpointStarted() splits into mutually exclusive in-recovery/normal branches. In recovery it only starts persisting (the persist window stays open) with the buffers collected by collectPreRecoveryBarrier, which walks the queue to the matching RecoveryCheckpointBarrier, retains pre-barrier data buffers and removes the sentinel; a missing barrier is an IOException wrapped as CheckpointException(CHECKPOINT_DECLINED). Mechanical constructor-caller adaptations ride along (UnknownInputChannel, LocalRecoveredInputChannel, InputChannelBuilder, benchmark factory), all passing needsRecovery=false at this stage. Includes a port of the FLINK-40016 double-persist regression test to the push-based recovery API.
Teach RemoteInputChannel to be created directly in a recovery state and to receive recovered buffers through the RecoverableInputChannel interface. The not-in-recovery checkpointStarted branch stays byte-identical to the previous behavior, and with needsRecovery=false the channel behaves exactly as before. - Constructor: drop ArrayDeque<Buffer> initialRecoveredBuffers, add needsRecovery (pre-completed stateConsumedFuture when false, same as LocalInputChannel). The BufferManager starts with credit notification disabled while in recovery (notifyInitiallyEnabled=!needsRecovery). - appendRecoveredBuffer() appends straight into receivedBuffers (NONE subpartition id, recovery sequence numbers from Integer.MIN_VALUE), so the consume path stays identical to the normal case. - recoveryEventStash (@GuardedBy receivedBuffers): ordinary upstream events arriving while credit is suppressed are stashed and replayed after the sentinel is consumed; live data buffers during recovery are a protocol violation asserted in onBuffer. - upstreamReady as CountDownLatch(1), counted down by the first onBuffer or by release; finishRecoveredBufferDelivery() awaits it and appends the EndOfFetchedChannelStateEvent sentinel. - onRecoveredStateConsumed(): unstash + bufferManager.enableNotify() (credit reopens) + complete stateConsumedFuture. - checkReadability() allows in-recovery reads before the partition request client is initialized. - checkpointStarted() in-recovery branch persists the buffers collected by collectPreRecoveryBarrier (walks receivedBuffers past the priority head to the matching RecoveryCheckpointBarrier, retains pre-barrier data buffers, removes the sentinel); only startPersisting is called, so the persist window stays open. A missing barrier declines the checkpoint (CHECKPOINT_DECLINED). - New public needsRecovery() getter, consumed by the netty stack in a later commit of this PR. Mechanical constructor-caller adaptations ride along (UnknownInputChannel, RemoteRecoveredInputChannel, InputChannelBuilder, benchmark factory, netty test channels), all passing needsRecovery=false. RemoteRecoveredInputChannel keeps calling remoteInputChannel.setup() until the conversion rework in the next commit moves setup() into RecoveredInputChannel#toInputChannel.
…ace; thread needsRecovery through gates RecoveredInputChannel#toInputChannel(boolean needsRecovery) replaces the no-arg variant; constructor migration of recovered buffers is retired: - needsRecovery=false: checkState(receivedBuffers.isEmpty()) + inputChannel.setup() (setup moves from RemoteRecoveredInputChannel into the conversion) + checkpointStopped(...). Flag-off conversion always happens after full consumption, so default behavior is unchanged. - needsRecovery=true (first used by the StreamTask recovery rework PR): create the physical channel in recovery state, then synchronously push every queued recovered data buffer via onRecoveredStateBuffer and append the EndOfFetchedChannelStateEvent sentinel (the legacy EndOfInputChannelStateEvent in the queue is dropped in translation). This is the simple in-memory drain -- FLINK-38544 transitional code, replaced by the disk drain when the spilling backend lands. The sentinel is appended directly rather than via finishRecoveredBufferDelivery(), which waits for upstream readiness that cannot arrive while the mailbox thread is still converting channels. - abstract toInputChannelInternal(ArrayDeque<Buffer>) becomes toInputChannelInternal(boolean). Gates thread the flag through: InputGate gains a concrete requestPartitions(boolean) (default overload ignores the flag); SingleInputGate implements requestPartitions(boolean) and convertRecoveredInputChannels(boolean) (no-arg kept as a @VisibleForTesting shim); getStateConsumedFuture() aggregation is unified on InputChannel#getStateConsumedFuture across all channels; UnionInputGate fans the flag out to member gates; InputGateWithMetrics delegates. Deliberate consequence: the base v1 flag-on StreamTask still converts at filtering-complete with unconsumed buffers, so the empty-queue checkState fires and the flag-on path is degraded until the StreamTask recovery rework lands -- covered by the randomization pin in the first commit of this PR. Default (flag-off) configuration is unaffected. Tests: RecoveredInputChannelTest is rewritten around the new conversion semantics (reject-while-unconsumed, empty-queue assert, and a transitional push-conversion case); Local/RemoteRecoveredInputChannelTest cover the empty-queue assert of the subclasses; UnionInputGateTest's helper adapts to the new toInputChannelInternal signature.
…tart view reader with zero credit Propagate the consumer channel's needsRecovery flag to the producer so it withholds credit while the consumer's exclusive buffers are on loan to recovery: - NettyMessage.PartitionRequest gains a needsRecovery boolean (ctor arg, writeBoolean/readBoolean, +Byte.BYTES in the length calculation). This is a wire-format change (+1 byte per PartitionRequest); safe under the usual single-version-cluster assumption for the TM network protocol, stated in the PR description. - CreditBasedSequenceNumberingViewReader ctor gains needsRecovery; numCreditsAvailable = needsRecovery ? 0 : initialCredit. - PartitionRequestServerHandler passes request.needsRecovery; NettyPartitionRequestClient passes inputChannel.needsRecovery(). Mechanical test adaptations: NettyMessageServerSideSerializationTest (round-trips the new field), PartitionRequestServerHandlerTest, PartitionRequestQueueTest, PartitionRequestRegistrationTest, CreditBasedSequenceNumberingViewReaderTest, CancelPartitionRequestTest, ServerTransportErrorHandlingTest.
…nnelStateEvent When the consume path polls the EndOfFetchedChannelStateEvent sentinel, assert the originating channel is a RecoverableInputChannel and call its onRecoveredStateConsumed(): the channel flips out of recovery, releases any upstream events held back during recovery and reopens the upstream so live data may flow again. The event itself is never delivered to the operator. The sentinel-consumption path is exercised end to end by the Local/RemoteInputChannel recovery tests of this PR (onRecoveredStateConsumed after polling the sentinel) and by the StreamTask recovery tests of the follow-up PRs.
…sync future chain Rewrite StreamTask channel-state recovery to run asynchronously on the channelIOExecutor and unify recovery completion on the gates' state-consumed futures. Read the during-recovery flag once via CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(jobConfig) and split restoreStateAndGates into recoverChannelsWithCheckpointing / recoverChannelsWithoutCheckpointing; recoveryCompletionFuture.whenComplete(-> mailboxProcessor.suspend()). Without checkpointing (flag off): submit readInputChannelState to the channelIOExecutor; convert and request partitions for each gate as soon as ITS OWN getStateConsumedFuture() completes (per-gate, avoiding the selective-reading multi-input deadlock); return completeAll(all stateConsumedFutures). Observable behavior equals master. With checkpointing (flag on, in-memory backend, transitional until the spilling backend lands): future chain runAsync(readInputChannelState, channelIOExecutor) -> requestPartitions(true) for all gates on the mailbox (conversion pushes the queued recovered buffers plus the sentinel) -> completeAll(the gates' stateConsumedFutures). An empty-input-gates short-circuit completes synchronously so a finite source cannot suspend the restore loop before recovery completes. restoreInternal now calls get() on the recovery future and rethrows the underlying cause, so recovery failures surface instead of being swallowed; the channelIOExecutor is shut down after the restore loop. Also move the gate-finishing loop (finishReadRecoveredState() for all gates) out of AbstractInputChannelRecoveredStateHandler.close() into StreamTask#readInputChannelState; close() only runs closeInternal() now. Adapt SequentialChannelStateReaderImplTest to call finishReadRecoveredState() itself and put TaskCheckpointingBehaviourTest under @timeout(120). Co-authored-by: Roman Khachatryan <khachatryan.roman@gmail.com>
…te future from the gate API Pure deletion of the now-unused v1 gating: the previous commit made StreamTask gate the flag-on RUNNING transition on the recovery future chain instead of the filtering-progress future, so the API has no callers left. Drop InputGate#getBufferFilteringCompleteFuture and its overrides in SingleInputGate, UnionInputGate and InputGateWithMetrics; IndexedInputGate#setCheckpointingDuringRecoveryEnabled / #isCheckpointingDuringRecoveryEnabled together with SingleInputGate's volatile backing field; and RecoveredInputChannel's bufferFilteringCompleteFuture field, getter and the finishReadRecoveredState lock-ordering commentary tied to it. The one remaining consumer of the flag -- the unbounded heap-buffer fallback in RecoveredInputChannel#requestBufferBlocking, which stays until disk spilling supersedes it -- now has the flag threaded explicitly from the job configuration through the recovery-handler call sites (requestBufferBlocking(boolean)) instead of reading it off the gate. Remove the corresponding testBufferFilteringCompleteFutureAggregation cases in SingleInputGateTest/UnionInputGateTest and adapt MockInputGate, MockIndexedInputGate, SingleInputGateBuilder and AlignedCheckpointsMassiveRandomTest.
A task that reaches END_OF_INPUT during recovery -- e.g. a bounded/batch
operator whose input is already fully available behind a blocking exchange, as
seen on TPC-H/TPC-DS HashAggregate -- previously suspended the restore mailbox
loop immediately. With checkpointing during recovery the recovery future
completes asynchronously, so this could end the restore loop before recovery
finished, tripping checkState(allGatesRecoveredFuture.isDone()) in
restoreInternal with "Mailbox loop interrupted before recovery was finished".
Input is intentionally processed during recovery (so checkpoint barriers can
arrive from inputs); the bug is only the premature lifecycle finish. When
END_OF_INPUT is reached while recovery is still in progress, suspend only the
default action (to avoid busy-spinning) and resume it once recovery completes,
via a new recoveryCompletionFuture field set in restoreStateAndGates.
processInput then re-fires END_OF_INPUT from the main mailbox loop and finishes
normally. Sources remain covered by the no-input-gates short-circuit.
Validated with a deterministic repro (bounded BATCH job, recovery window
widened): the race reproduces with this deferral disabled and is gone with it
enabled. Full regression green: heavy-deployment e2e 6/6; ChainOrderTest,
KeyedComplexChainTest, KeyedPartitionWindowedStreamITCase; UnalignedCheckpoint
{,Rescale}ITCase (the with-recovered-state path) 11/11 and 50/50.
This was referenced Jul 6, 2026
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR depends on #28651 and #28652
Please only review the last 3 commits (prefixed [FLINK-39522]).
What is the purpose of the change
[FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes.
It's part 3 of FLINK-38544 (checkpointing during recovery of unaligned checkpoint state). StreamTask channel-state recovery is restructured into an asynchronous future chain on the channelIOExecutor, and recovery completion is unified on the gates' state-consumed futures. With checkpointing-during-recovery enabled, the task now consumes recovered state only after rescale filtering has fully completed (previously consumption overlapped with filtering); the filtered buffers are held in the recovered channels' memory queues and handed to the physical channels through the push interface at conversion. The now-unused filtering-progress gate API is removed. Non-CDR observable behavior is preserved; the defer-finish commit fixes an existing race where a finite task could finish before recovery completed. Checkpoints during recovery are declined at this stage.
Brief change log
Verifying this change
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation