rkhachatryan commented on code in PR #29189:
URL: https://github.com/apache/flink/pull/29189#discussion_r4015931603
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java:
##########
@@ -133,7 +133,7 @@ public void filterAndRewrite(
+ ". This gate is not a network input and should
not have recovered buffers.");
}
gateHandler.filterAndRewrite(
- oldSubtaskIndex, oldChannelIndex, sourceBuffer,
outputSerializer);
+ oldSubtaskIndex, oldChannelIndex, sourceBuffer.retainBuffer(),
outputSerializer);
Review Comment:
**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.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java:
##########
@@ -592,7 +592,7 @@ public void recover(
channelInfo.getGateIdx(),
oldSubtaskIndex,
channelInfo.getInputChannelIdx(),
- buffer.retainBuffer(),
+ buffer,
Review Comment:
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.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java:
##########
@@ -351,6 +351,27 @@ void testPreFilterSegmentFreedOnClose() throws Exception {
assertThat(filteringHandler.getPreFilterSegmentForTesting()).isNull();
}
+ @Test
+ void testPreFilterBufferRecycledWhenFilterAndRewriteThrows() throws
Exception {
Review Comment:
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.
--
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]
