Zakelly commented on code in PR #24933: URL: https://github.com/apache/flink/pull/24933#discussion_r1639554722
########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java: ########## @@ -153,6 +152,22 @@ public abstract class FileMergingSnapshotManagerBase implements FileMergingSnaps /** The current space statistic, updated on file creation/deletion. */ protected SpaceStat spaceStat; + /** + * This map records shared state dirs which need be clean up when the FileMergingSnapshotManager + * close. The key is SubtaskKey the shared state dir belong to, and the value is the count of + * the ongoing checkpoint which reference the dir. If a checkpoint which reference the shared + * dir complete, the corresponding shared dir will be removed from this map, because the + * ownership is transferred to JobManager. + */ + private final Map<SubtaskKey, Long> sharedDirToCleanRef = new ConcurrentHashMap<>(); Review Comment: I'd suggest a new class that bundle `DirectoryStreamStateHandle` and its reference count. ########## flink-runtime/src/main/java/org/apache/flink/runtime/state/filemerging/DirectoryStreamStateHandle.java: ########## @@ -19,23 +19,31 @@ package org.apache.flink.runtime.state.filemerging; import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.state.DirectoryStateHandle; import org.apache.flink.runtime.state.PhysicalStateHandleID; import org.apache.flink.runtime.state.SharedStateRegistryKey; import org.apache.flink.runtime.state.StreamStateHandle; import javax.annotation.Nonnull; -import java.nio.file.Path; import java.util.Optional; /** Wrap {@link DirectoryStateHandle} to a {@link StreamStateHandle}. */ -public class DirectoryStreamStateHandle extends DirectoryStateHandle implements StreamStateHandle { +public class DirectoryStreamStateHandle implements StreamStateHandle { private static final long serialVersionUID = 1L; - public DirectoryStreamStateHandle(@Nonnull Path directory, long directorySize) { - super(directory, directorySize); + /** The path that describes the directory, as a string, to be serializable. */ + private final String directoryString; + + /** Transient path cache, to avoid re-parsing the string. */ + private transient Path directory; Review Comment: Seems unnecessary? I mean we could just hold the `Path` with `final` keyword, and no need for the `String directoryString`? ########## flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsMergingCheckpointStorageLocation.java: ########## @@ -80,6 +81,12 @@ public FsMergingCheckpointStorageLocation( reference, fileStateSizeThreshold, writeBufferSize); + + // Record file-merging managed dir reference when FsMergingCheckpointStorageLocation create. + if (fileMergingSnapshotManager instanceof FileMergingSnapshotManagerBase) { + ((FileMergingSnapshotManagerBase) fileMergingSnapshotManager) + .recordManagedDirReference(subtaskKey, checkpointId); Review Comment: How about making this as an interface of `FileMergingSnapshotManager`? And how about a new name `retainManagedDirectory` ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java: ########## @@ -496,14 +497,34 @@ protected void discardCheckpoint(long checkpointId) throws IOException { // Checkpoint Listener // ------------------------------------------------------------------------ + /** + * {@link FsMergingCheckpointStorageLocation} use this method let the file merging manager know + * an ongoing checkpoint may reference the managed dirs. + */ + public void notifyCheckpointStart(SubtaskKey subtaskKey, long checkpointId) { + managedSharedStateDirHandles + .getOrDefault(subtaskKey, NON_HANDLE_INSTANCE) + .increaseRefCountWhenCheckpointStart(checkpointId); Review Comment: use `computeIfPresent` ? ########## flink-runtime/src/main/java/org/apache/flink/runtime/state/filemerging/DirectoryStreamStateHandle.java: ########## @@ -19,23 +19,26 @@ package org.apache.flink.runtime.state.filemerging; import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.state.DirectoryStateHandle; import org.apache.flink.runtime.state.PhysicalStateHandleID; import org.apache.flink.runtime.state.SharedStateRegistryKey; import org.apache.flink.runtime.state.StreamStateHandle; import javax.annotation.Nonnull; -import java.nio.file.Path; import java.util.Optional; /** Wrap {@link DirectoryStateHandle} to a {@link StreamStateHandle}. */ Review Comment: Maybe you should change this javadoc ########## flink-runtime/src/main/java/org/apache/flink/runtime/state/filemerging/DirectoryStreamStateHandle.java: ########## @@ -85,6 +92,18 @@ public String toString() { * @return DirectoryStreamStateHandle with zero size. */ public static DirectoryStreamStateHandle forPathWithZeroSize(@Nonnull Path directory) { Review Comment: How about naming it `of`? ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java: ########## @@ -496,14 +497,34 @@ protected void discardCheckpoint(long checkpointId) throws IOException { // Checkpoint Listener // ------------------------------------------------------------------------ + /** + * {@link FsMergingCheckpointStorageLocation} use this method let the file merging manager know + * an ongoing checkpoint may reference the managed dirs. + */ + public void notifyCheckpointStart(SubtaskKey subtaskKey, long checkpointId) { Review Comment: I'd suggest this to be an interface of `FileMergingSnapshotManager`. -- 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