[FLINK-39524][checkpoint] Fetched channel state: spill files, filtered write, reader#28661
Open
1996fanrui wants to merge 25 commits into
Open
[FLINK-39524][checkpoint] Fetched channel state: spill files, filtered write, reader#286611996fanrui wants to merge 25 commits into
1996fanrui wants to merge 25 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.
…_OP and a barrier-inserting in-memory implementation New RecoveryCheckpointTrigger (checkpoint.channel): void snapshotAndInsertBarriers(long checkpointId) -- narrow signature, widened to return a snapshot reader when the spilling backend lands; NO_OP singleton (no-op); NOT_READY singleton (throws CheckpointException(CHECKPOINT_DECLINED_TASK_NOT_READY) -- transient, the coordinator retries). New transitional in-memory implementation InMemoryRecoveryCheckpointTrigger: holds the task's List<RecoverableInputChannel>; snapshotAndInsertBarriers = forEach(ch -> ch.insertRecoveryCheckpointBarrierIfInRecovery(cpId)). The javadoc states why no snapshot exists (one-shot push means no undrained residue) and that the disk drainer replaces it. Deleted when the spilling backend lands. Unit tests for all three implementations. Co-authored-by: Roman Khachatryan <khachatryan.roman@gmail.com>
…ough the recovery trigger ChannelState gains a RecoveryCheckpointTrigger (the legacy 1-arg ctor defaults to NO_OP); new onCheckpointStartedForAllInputs(CheckpointBarrier): (1) trigger.snapshotAndInsertBarriers(cpId); (2) for (input : inputs) input.checkpointStarted(barrier). CheckpointException is rethrown as-is (routes to checkpoint abort, not task failure); other IOException via rethrowIOException. (Step 3 -- spilled-slice replay through the channel-state writer -- is added when the spilling backend lands.) AlternatingCollectingBarriers and AlternatingWaitingForFirstBarrierUnaligned: replace the inline per-input checkpointStarted loop with state.onCheckpointStartedForAllInputs(...) (behaviorally inert with NO_OP). Tests: ChannelStateDispatcherTest (2-step scope), AlternatingCollectingBarriersDispatchHookTest, AlternatingWaitingForFirstBarrierUnalignedDispatchHookTest.
… and the StreamTask lifecycle InputProcessorUtil.createCheckpointBarrierHandler: overload taking a RecoveryCheckpointTrigger (the old signature is preserved as a NO_OP delegator); SingleCheckpointBarrierHandler.unaligned/alternating gain the trigger param (aligned keeps the NO_OP default through ChannelState's 1-arg ctor); OneInputStreamTask/TwoInputStreamTask/MultipleInputStreamTask pass StreamTask.getRecoveryCheckpointTrigger(). (The channel-state writer is threaded through the same seams when the spilling backend lands.) StreamTask: recoveryCheckpointTrigger field (starts NOT_READY) + mailbox-thread-asserting getter lambda; all asynchronous mutations via setRecoveryCheckpointTrigger (a mailbox mail). Lifecycle inside recoverChannelsWithCheckpointing: NOT_READY before/during filtering (checkpoints declined as task-not-ready; transient, the coordinator retries) -> install the in-memory barrier-inserting trigger at conversion, composing on requestPartitions' returned List<RecoverableInputChannel> -> swap to NO_OP when completeAll(gates' stateConsumedFutures) completes (gate on the futures, not on push completion; the mailbox-mail gap is safe because a barrier-inserting trigger with no in-recovery channels behaves as NO_OP). recoverChannelsWithoutCheckpointing and the empty-input-gates short-circuit go straight to NO_OP. Also fix the transitional conversion push's sentinel timing: the EndOfFetchedChannelStateEvent sentinel is no longer appended inside toInputChannelInRecovery() (where it becomes consumable before the upstream connection exists, letting a spurious post-recovery poll hit a LocalInputChannel without a subpartition view: "Queried for a buffer before requesting the subpartition"). Instead the recovery chain appends it via finishRecoveredBufferDelivery() on the channelIOExecutor after partitions are requested -- the method waits for upstream readiness, restoring the invariant the disk drainer also relies on, and mirroring where the drainer will run. Tests: TestBarrierHandlerFactory adaptation; StreamTask trigger-lifecycle source-level invariants (transitional, replaced with the disk drainer wiring when the spilling backend lands); RecoveredInputChannelTest push-conversion case adapted to the sentinel timing.
…able filtering ITCase TestStreamEnvironment: restore randomize(conf, CHECKPOINTING_DURING_RECOVERY_ENABLED, true, false), closing the transitional randomization window opened when the v1 conversion-migration path was retired. Re-enable RecoveredStateFilteringLargeRecordITCase. From here on the randomized ITCase fleet runs flag-on ~50% against the in-memory backend (same memory profile as base v1, which ran the same randomization with the same heap fallback). The spilling PRs must keep the fleet green -- this is the acceptance bar for "lightweight backend replacement".
Sealed container over an ordered List<Path> of spill files with an acquire()/release() ref-counted lifecycle that deletes the files when the last grant is released (cleanedUp guard); close() forces cleanup. Deviation from the plan's commit scope: the reader() entry point and FetchedChannelStateSnapshot are deferred to the forward-only-reader commit of this PR — both hard-depend on FetchedChannelStateReader / FetchedChannelStateReaderImpl.Position, which do not exist yet. Everything present here is byte-identical to its final form. FetchedChannelStateRefCountTest is likewise deferred (it needs the TestSpillWriter from the spill-writing-handlers commit).
…ormat New AbstractSpillingHandler (in RecoveredChannelStateHandler.java, extending the abstract input-channel base): a reusable DataOutputSerializer accumulates one channel's segment [gateIdx][channelIdx][bufferLength][body...]; the body length is backfilled via writeIntUnsafe at seal; flush through OffsetAwareOutputStream; 64 MiB soft rotation; closeInternal() seals, closes the stream, and builds the FetchedChannelState handoff (calling acquire()). New SpillingNoFilteringHandler writes the recovered buffer's bytes verbatim via segmentSerializerFor(...).write(memorySegment, offset, len). flink-core OffsetAwareOutputStream: ctor package-private -> public so flink-runtime can construct it. Also adds the @nullable FetchedChannelState getProducedChannelState() hook to AbstractInputChannelRecoveredStateHandler (returns null; the spilling base overrides it) and restores the final class javadoc on the abstract base. The factory is NOT switched: no production path selects a spilling handler yet. Transitional deviation: AbstractSpillingHandler passes the base's transitional third ctor arg (true); reverts to the 2-arg base ctor when the spilling backend lands. FetchedChannelStateRefCountTest rides here (it needs TestSpillWriter) minus testReaderAcquiresAndReleasesOnClose, which follows with the reader commit.
…t into a spill segment filterAndRewrite now appends each surviving record (length-prefixed via a 4B placeholder + writeIntUnsafe backfill) into a caller-provided DataOutputSerializer: no network buffers, no InterruptedException. The deserializer creation is inlined (always filterContext.getTmpDirectories(); the java.io.tmpdir fallback is gone). New SpillingWithFilteringHandler: getBuffer() is the reusable heap pre-filter-segment logic (verbatim from the v1 FilteringHandler); recover() routes filter output into the spill segment of the mapped channel. The v1 FilteringHandler stays present and factory-selected -- it is still the memory backend. Since it can no longer call the rewritten method, the old buffer-delivering filter loop (BufferSupplier, List<Buffer> overloads, the chunking machinery and its two GateFilterHandler fields) is kept as clearly marked FLINK-38544-transitional overloads; they die together with the class when the spilling backend lands. RecoveredStateFilteringLargeRecordITCase (explicitly flag-on, v1 path) still passes after this commit. AbstractInputChannelRecoveredStateHandler.create(...) gains the final String[] spillTmpDirectories parameter (unused by the transitional in-memory branches); SequentialChannelStateReaderImpl passes filterContext.getTmpDirectories() mechanically. Tests: ChannelStateFilteringHandlerTest, GateFilterHandlerTest and GateFilterHandlerBufferOwnershipTest (rewritten against the serializer sink), RecoveredChannelStateHandlerFilterRoutingTest, and the InputChannelRecoveredStateHandlerTest adaptation. The three helpers that in final form obtain spilling handlers through the factory construct them directly for now (transitional; the factory-switch commit restores the factory calls).
FetchedChannelStateReader interface: nextSegment(), snapshot(), emptyReader(); inner SpillSegment (channelInfo(), bodyStream(), length(), commit()). FetchedChannelStateReaderImpl: sequential file IO over the spill files, a bounded per-segment body InputStream, current/committed positions, and snapshot resume with skip-only-on-first-positioning. Completes the pieces deferred from the container commit for dependency reasons: FetchedChannelStateSnapshot (the immutable one-shot resume point holding one lifecycle grant and a Position; reader() at most once, fail-loud), FetchedChannelState#reader(), and FetchedChannelStateRefCountTest#testReaderAcquiresAndReleasesOnClose. All files in this commit are byte-identical to their final form.
…returns Optional<FetchedChannelState> Interface and NO_OP return Optional.empty(). SequentialChannelStateReaderImpl reworks the close ordering (nested try-with-resources so the produced spill file is published only after stateHandler.close() has flushed the filter writer) and returns Optional.ofNullable(handler.getProducedChannelState()). The original branch's dead producedChannelState field is intentionally not re-added. StreamTask#readInputChannelState asserts the memory-backend invariant on the new return value: checkState(reader.readInputData(...).isEmpty()) -- the factory still selects the in-memory handlers, so no production path produces a FetchedChannelState until the backend switch.
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, #28652, #28659 and #28660
Please only review the last 5 commits (prefixed [FLINK-39524]).
What is the purpose of the change
[FLINK-39524] Fetched channel state: spill files, filtered write, reader.
It's part 5 of FLINK-38544 (checkpointing during recovery of unaligned checkpoint state). This PR introduces the on-disk subsystem for recovered channel state: an append-only segmented spill-file format written by the recovery read path (optionally through the rescale record filter), a ref-counted file container with snapshot/resume semantics, and a forward-only reader. Runtime behavior is unchanged: the handler factory does not select the new handlers, so no production path produces or consumes spill files.
Brief change log
Verifying this change
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation