[FLINK-39520] Preparatory refactoring for channel state recovery (no behavior change)#28651
Open
1996fanrui wants to merge 5 commits into
Open
[FLINK-39520] Preparatory refactoring for channel state recovery (no behavior change)#286511996fanrui wants to merge 5 commits into
1996fanrui wants to merge 5 commits into
Conversation
Collaborator
…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.
This was referenced Jul 6, 2026
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.
What is the purpose of the change
[FLINK-39520] Preparatory refactoring for channel state recovery (no behavior change).
It's part 1 of FLINK-38544 (checkpointing during recovery of unaligned checkpoint state). This PR only refactors code in preparation for the reworked channel-state recovery; it also carries one standalone
[hotfix]commit fixing a busy-spin in the multiple-input processor.Brief change log
Verifying this change
This change is a refactoring without behavior change; it is covered by existing tests (mechanical test adaptations only).
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation