[FLINK-40667][network] Fix buffer ref-count leak in recovery channel-state filtering

[1996fanrui]

What is the purpose of the change

During checkpointing-during-recovery, SpillingWithFilteringHandler.recover() retains the
pre-filter buffer at the call site, before segmentSerializerFor() and the
ChannelStateFilteringHandler dispatcher's early throws (invalid gateIndex / null gate
handler) run. Any of those throws orphans the retained reference, so its refCount never
reaches 0 and closeInternal() later free()s a MemorySegment still wrapped by a live
NetworkBuffer.

Brief change log

• recover() passes the base buffer instead of buffer.retainBuffer().
• The retain now happens where the dispatcher delegates to the inner GateFilterHandler,
  whose sourceBufferOwnershipTransferred guard already recycles on throw.

Verifying this change

Added InputChannelRecoveredStateHandlerTest#testPreFilterBufferRecycledWhenFilterAndRewriteThrows:
asserts the pre-filter buffer is recycled when filterAndRewrite throws. Fails before the fix,
passes after.

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
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no

[flinkbot]

CI report:

• b26675e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[rkhachatryan]

AI-generated review — Claude Opus 5 via Claude Code /code-review (default effort). Findings were manually verified against the source at 61f4d35; treat as input, not a verdict.

The diagnosis looks right: Java evaluates buffer.retainBuffer() before segmentSerializerFor(getMappedChannels(...)) and before the dispatcher's two IllegalStateExceptions, so any of those throws leaves refCnt at 2, the finally drops it only to 1, the recycler never fires, preFilterBufferInUse stays true, and closeInternal() later free()s a segment still wrapped by a live NetworkBuffer. Real bug, right place to fix it.

Two things to address before merge (inline), plus two non-blocking notes:

1. The change breaks the existing GateFilterHandlerBufferOwnershipTest#testCloseRecyclesDeserializerHeldBufferAfterError.
2. The two filterAndRewrite overloads end up with opposite ownership contracts, undocumented.
3. (non-blocking, pre-existing) The free()-under-live-buffer hazard survives on the post-transfer throw path.
4. (non-blocking) The new test exercises a branch that can't occur in production.

[rkhachatryan]

Non-blocking, and pre-existing rather than introduced here: the same free()-under-live-buffer hazard survives on the post-transfer throw path.

If filterAndRewrite throws after vc.setNextBuffer (e.g. a DataOutputSerializer IOException in emitAggregated), the deserializer still holds the reference, this finally goes 2 → 1, and preFilterBufferInUse stays true. In SequentialChannelStateReaderImpl#readInputData the inner try-with-resources closes stateHandler first, so closeInternal() calls preFilterSegment.free() while the deserializer's NetworkBuffer is still live; the outer filteringHandler.close() only clears it afterwards.

Harmless today (the task is already failing and nothing reads the freed segment), but it is the same bug class this PR closes. Either swap the close order, or have closeInternal() refuse to free while preFilterBufferInUse. Fine as a follow-up.

[1996fanrui]

Agreed. Ownership stays intact here — the buffer is still recycled by close() → clear(); it's only a free-vs-recycle ordering window on the already-failing teardown path, with no leak or lasting effect. Leaving it as a follow-up.

[1996fanrui]

Thanks @rkhachatryan for the review, all comments are addressed or replied.

[rkhachatryan]

Thanks for the PR, LGTM