fredia commented on code in PR #24806:
URL: https://github.com/apache/flink/pull/24806#discussion_r1606322375
##########
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:
Should the corresponding checkpoint also be removed from
`notifiedSubtaskCheckpoint` ?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/SubtaskFileMergingManagerRestoreOperation.java:
##########
@@ -93,8 +94,12 @@ public void restore() {
Stream<SegmentFileStateHandle> segmentStateHandles =
Stream.of(keyedStateHandles, operatorStateHandles)
.flatMap(Function.identity())
- .filter(handle -> handle instanceof
SegmentFileStateHandle)
+ .filter(
+ handle ->
+ (handle instanceof
SegmentFileStateHandle)
+ && !(handle instanceof
EmptySegmentFileStateHandle))
.map(handle -> (SegmentFileStateHandle) handle);
+ System.out.println("Restoring from checkpoint " + checkpointId);
Review Comment:
LOG.info()?
--
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
