[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:

• 61f4d35 Azure: FAILURE

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]

This breaks GateFilterHandlerBufferOwnershipTest#testCloseRecyclesDeserializerHeldBufferAfterError (not updated in this PR).

That test calls the dispatcher 5-arg overload with a buffer at refCnt 1 and asserts sourceBuffer.isRecycled() after the close() chain. With the added retain: 1 → 2; the serializer throws after vc.setNextBuffer, so sourceBufferOwnershipTransferred is true and GateFilterHandler's catch correctly skips recycling; then close() → clear() → deserializer.clear() brings it 2 → 1. NetworkBuffer.isRecycled() is refCnt() == 0, so the assertion fails.

Related: the two overloads now have opposite ownership contracts. GateFilterHandler.filterAndRewrite (inner) consumes the caller's reference — recycles on pre-transfer throw, hands ownership to the deserializer otherwise. After this change the dispatcher does not; it adds its own. Same method name, same package, inverted semantics, and neither javadoc says so. The broken test above is exactly the failure mode this invites.

Alternative that avoids the asymmetry: keep buffer.retainBuffer() in recover(), but hoist segmentSerializerFor(...) into a local before it, and make the dispatcher's two early throws recycle sourceBuffer — matching the inner contract. Both overloads then behave identically and the existing ownership test stays green. If you prefer the current shape, the test needs updating and both javadocs should state the contract explicitly.

[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.

[rkhachatryan]

Non-blocking: this covers the one branch that can't occur in production. The zero-length GateFilterHandler[] hits gateIndex >= gateHandlers.length, but createFromContext always sizes the array to inputGates.length, so channelInfo.getGateIdx() is never out of range in a real job.

The two reachable pre-retain throw sites named in the PR description are untested: segmentSerializerFor(...) / getMappedChannels(...) failing during argument evaluation, and gateHandler == null (reachable — createGateHandler returns null for a gate with no virtual channels). A non-empty array holding a null entry would exercise the actual production path.