Zakelly commented on code in PR #24806:
URL: https://github.com/apache/flink/pull/24806#discussion_r1606346053
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java:
##########
@@ -485,6 +507,36 @@ public void notifyCheckpointSubsumed(SubtaskKey
subtaskKey, long checkpointId)
}
}
}
+ notifyReleaseCheckpoint(subtaskKey, checkpointId);
+ }
+
+ private void notifyReleaseCheckpoint(SubtaskKey subtaskKey, long
checkpointId)
+ throws IOException {
+ synchronized (notifyLock) {
+ if (notifiedCheckpoint.contains(checkpointId)) {
+ // already release, skip
+ return;
+ }
+ Set<SubtaskKey> knownSubtask =
+ notifiedSubtaskCheckpoint.computeIfAbsent(checkpointId,
(e) -> new HashSet<>());
+ knownSubtask.add(subtaskKey);
+ if (knownSubtask.containsAll(managedSharedStateDir.keySet())) {
+ // all known subtask has been notified.
+ tryDiscardCheckpoint(checkpointId);
+ }
+ }
+ }
+
+ private void tryDiscardCheckpoint(long checkpointId) throws IOException {
+ synchronized (notifyLock) {
+ if (!notifiedCheckpoint.contains(checkpointId)) {
+ notifiedCheckpoint.add(checkpointId);
+ discardCheckpoint(checkpointId);
+ if (notifiedCheckpoint.size() > NUM_GHOST_CHECKPOINT_IDS) {
+ notifiedCheckpoint.pollFirst();
Review Comment:
Added in `notifyReleaseCheckpoint` and `tryDiscardCheckpoint`
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
