1996fanrui commented on code in PR #27783:
URL: https://github.com/apache/flink/pull/27783#discussion_r3016090127
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java:
##########
@@ -147,29 +124,35 @@ public void recover(
}
}
} finally {
- if (!buffer.isRecycled()) {
- buffer.recycleBuffer();
- }
+ buffer.recycleBuffer();
Review Comment:
Good point.
I tried it but found it gets worse, several calls here could throw
(`getMappedChannels`, `EventSerializer.toBuffer`, `filterAndRewrite`). Without
retain, we'd need explicit ownership tracking:
```java
@Override
public void recover(
InputChannelInfo channelInfo,
int oldSubtaskIndex,
BufferWithContext<Buffer> bufferWithContext)
throws IOException, InterruptedException {
Buffer buffer = bufferWithContext.context;
if (buffer.readableBytes() == 0) {
buffer.recycleBuffer();
return;
}
boolean ownershipTransferred = false;
try {
RecoveredInputChannel channel = getMappedChannels(channelInfo);
if (filteringHandler != null) {
ownershipTransferred = true;
recoverWithFiltering(channel, channelInfo, oldSubtaskIndex,
buffer);
} else {
channel.onRecoveredStateBuffer(
EventSerializer.toBuffer(
new SubtaskConnectionDescriptor(
oldSubtaskIndex,
channelInfo.getInputChannelIdx()),
false));
ownershipTransferred = true;
channel.onRecoveredStateBuffer(buffer);
}
} catch (Throwable t) {
if (!ownershipTransferred) {
buffer.recycleBuffer();
}
throw t;
}
}
```
The retain/recycle pair costs two atomic int ops but gives us one
unconditional `finally` for all failure paths — simpler and safer. Keeping
as-is.
--
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]
