1996fanrui commented on code in PR #20233:
URL: https://github.com/apache/flink/pull/20233#discussion_r923185089
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java:
##########
@@ -60,89 +63,78 @@
public class ChannelStateWriterImpl implements ChannelStateWriter {
private static final Logger LOG =
LoggerFactory.getLogger(ChannelStateWriterImpl.class);
- private static final int DEFAULT_MAX_CHECKPOINTS =
- 1000; // includes max-concurrent-checkpoints + checkpoints to be
aborted (scheduled via
- // mailbox)
Review Comment:
Hi @pnowojski
From the comment, we can get that the `ConcurrentMap<Long,
ChannelStateWriteResult> results;` saves ongoingCheckpoints and
abortedCheckpoints.
But the checkpoints that were aborted before start were not saved. We want
aborted checkpoints to ignore start. So I introduced the ` NavigableSet<Long>
abortedCheckpointIds;` to save all aborted checkpointIds.
- If checkpointId is aborted before start, it will be saved into this set.
- When the start is called, if checkpointId in this set, the start should be
ignored.
And introduced the `long ongoingCheckpointId;` to save the ongoing
checkpointId. Because the concurrent unaligned checkpoint isn't supported, so
the `long` is enough. And checkpoints with checkpointId less than
`ongoingCheckpointId` should be aborted. So I remove these checkpointIds when
the ongoingCheckpointId is updated.
What do you think?
--
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]
