1996fanrui opened a new pull request, #28001:
URL: https://github.com/apache/flink/pull/28001
## 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**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
