fredia commented on code in PR #24640:
URL: https://github.com/apache/flink/pull/24640#discussion_r1557536559
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java:
##########
@@ -547,4 +548,43 @@ private void createManagedDirectory(Path managedPath) {
@Override
public void close() throws IOException {}
+
+ // ------------------------------------------------------------------------
+ // restore
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void restoreStateHandles(
+ long checkpointId, SubtaskKey subtaskKey,
Stream<SegmentFileStateHandle> stateHandles) {
+
+ Set<LogicalFile> uploadedLogicalFiles;
+ synchronized (lock) {
+ uploadedLogicalFiles =
+ uploadedStates.computeIfAbsent(checkpointId, id -> new
HashSet<>());
+ }
+
+ stateHandles.forEach(
+ fileHandle -> {
+ // The file cleanup ownership of recovered physicalFile
belongs to
+ // jobManager, so we should set file deleter to "null" for
them.
Review Comment:
IIUC, JM is only aware of `SegmentFileStateHandle`, and the
`SegmentFileStateHandle#discardState()` is empty(). Currently, the deletion of
physicalFile relies on JM's checkpoint completed/abort notification, which is
executed on the TM. So, the file deleter shouldn't be null.
--
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
