[FLINK-39519][checkpoint] Allocate pre-filter source buffers from a reusable heap segment

[1996fanrui]

What is the purpose of the change

Under FLINK-39519, the channel-state recovery loop can deadlock when checkpointing-during-recovery filters inflight buffers: the single-threaded recovery thread first requests buffers to read pre-filter state, then requests more buffers to write post-filter output. If pre-filter buffers exhaust the Network Buffer Pool, post-filter allocation blocks and stalls the thread, so pre-filter buffers can never be consumed and released.

This PR isolates the pre-filter buffer from the Network Buffer Pool by allocating it from a dedicated heap segment.

Brief change log

• InputChannelRecoveredStateHandler lazily allocates one heap MemorySegment per task (sized to memorySegmentSize), reuses it across every pre-filter getBuffer() call, and frees it in close(). Non-filtering mode is unchanged.
• A runtime check (!preFilterBufferInUse) enforces the one-buffer-at-a-time invariant. The custom BufferRecycler flips the flag back on recycle without freeing the segment. Any future regression of the invariant fails loudly with IllegalStateException instead of silently corrupting memory.
• Wiring:
  • RecordFilterContext gains a required memorySegmentSize parameter (with checkArgument > 0) and a getMemorySegmentSize() accessor. disabled() factory uses MemoryManager.DEFAULT_PAGE_SIZE.
  • StreamTask.createRecordFilterContext() passes ConfigurationParserUtils.getPageSize(jobConfiguration).
  • SequentialChannelStateReaderImpl passes filterContext.getMemorySegmentSize() to the handler constructor.
• Added CommonTestUtils.waitForNCheckpointsWithInflightBuffers to wait for N completed checkpoints that carry inflight channel state.

Why a single reusable segment is enough

At most one pre-filter source buffer is in flight per task at any moment, guaranteed by Flink's existing recovery design:

1. ChannelStateChunkReader.readChunk() is a single-threaded while-loop (getBuffer → fill → recover(finally recycle)). The next getBuffer() can only start after recover() returns — structural concurrency = 1.
2. SpillingAdaptiveSpanningRecordDeserializer.getNextRecord() recycles the current buffer the instant isBufferConsumed=true, which happens on both PARTIAL_RECORD (record spans buffers) and LAST_RECORD_FROM_BUFFER (buffer exhausted). filterAndRewrite() breaks its loop on the same condition.
3. Cross-buffer bytes are always copied out before recycle: SpanningWrapper.transferFrom / addNextChunkFromMemorySegment copy partial bytes into an internal byte[] (or into SpanningWrapper's own spill file for records ≥ 5 MB). The only retained segment pointer (leftOverData / NonSpanningWrapper.segment) is read via segment.get(...) — copies into the target — and only while isBufferConsumed=false.
4. Exception paths converge through ChannelStateFilteringHandler.close() → VirtualChannel.clear() → deserializer.clear(), which recycles any buffer the deserializer still holds.

Verifying this change

Unit tests (InputChannelRecoveredStateHandlerTest)

• testPreFilterBufferIsolationFromNetworkBufferPool: filtering-mode getBuffer() returns a heap-backed buffer; the Network Buffer Pool is untouched.
• testNonFilteringModeUsesNetworkBufferPool: non-filtering path preserved.
• testPreFilterSegmentReusedAcrossCalls: successive getBuffer()/recycle cycles return the same MemorySegment instance.
• testGetBufferThrowsWhenPriorBufferNotRecycled: runtime invariant check throws IllegalStateException.
• testPreFilterSegmentFreedOnClose: segment freed on handler close.
• testMemorySegmentSizeExposedAndValidated: context validation and getter.

Integration test (RecoveredStateFilteringLargeRecordITCase)

Pins the network memory segment to Flink's minimum page size and generates records 8× the segment size so each record spans many consecutive source buffers. A sleeping downstream map induces back-pressure on the keyBy shuffle so inflight buffers actually accumulate. The test runs three phases with random parallelism per phase:

1. Phase 1: take an unaligned checkpoint with inflight buffers.
2. Phase 2: restore from phase 1 and wait for one post-recovery checkpoint with inflight buffers. That checkpoint — the one taken during recovery itself — becomes phase 3's restore target, so phase 3 exercises the "recovery-of-a-recovered-checkpoint" path.
3. Phase 3: restore from phase 2 and wait for five post-recovery checkpoints to shake out stability issues beyond the first post-recovery checkpoint.

Both filtering flags (UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM and CHECKPOINTING_DURING_RECOVERY_ENABLED) are pinned on so every run deterministically exercises the filtering handler.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no (only the recovery path)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — channel state recovery with filtering
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no (bug fix / robustness improvement)
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 0095cb3 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build