1996fanrui commented on code in PR #21131:
URL: https://github.com/apache/flink/pull/21131#discussion_r1008188410
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestDispatcherImpl.java:
##########
@@ -98,18 +120,19 @@ private ChannelStateCheckpointWriter
buildWriter(CheckpointStartRequest request)
streamFactoryResolver.resolveCheckpointStorageLocation(
request.getCheckpointId(),
request.getLocationReference()),
serializer,
- () -> writers.remove(request.getCheckpointId()));
+ () -> this.writer = null);
Review Comment:
Thanks @pnowojski help review.
Good catch, I have added the `checkState` here. After I think about it, I
don't think it can happen.
The call stack of this callback:
`ChannelStateWriteRequestDispatcherImpl#dispatchInternal` ->
`ChannelStateCheckpointWriter(writer)#completeInput/completeOutput` ->
`complete` -> `doComplete.run()` -> `callback` ->
`ChannelStateWriteRequestDispatcherImpl.writer = null`
First of all, here is single thread. `writer#completeInput/completeOutput`
will clean up itself.
If there is a new writer in `ChannelStateWriteRequestDispatcherImpl`, the
old writer's callback will not be called. In other words: only the current
writer will call the callback and clean itself up.
Please correct me if anything is wrong, thanks~
--
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]
