[FLINK-32086][checkpointing] Fix deletion fail of file-merging manage…

…d dir
Gerrrr · Jun 18, 2024 · 9f90e58 · 9f90e58
1 parent 6723ad7
commit 9f90e58
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 37 deletions.
diff --git a/.../java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java b/.../java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java
@@ -42,7 +42,6 @@
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.GuardedBy;
 
-import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Collection;
@@ -210,8 +209,7 @@ public void initFileSystem(
             createManagedDirectory(managedExclusivePath);
             this.managedExclusiveStateDir = managedExclusivePath;
             this.managedExclusiveStateDirHandle =
-                    DirectoryStreamStateHandle.forPathWithZeroSize(
-                            new File(managedExclusivePath.getPath()).toPath());
+                    DirectoryStreamStateHandle.of(managedExclusivePath);
             this.writeBufferSize = writeBufferSize;
         }
     }
@@ -224,9 +222,7 @@ public void registerSubtaskForSharedStates(SubtaskKey subtaskKey) {
             createManagedDirectory(managedPath);
             managedSharedStateDir.put(subtaskKey, managedPath);
             managedSharedStateDirHandles.put(
-                    subtaskKey,
-                    DirectoryStreamStateHandle.forPathWithZeroSize(
-                            new File(managedPath.getPath()).toPath()));
+                    subtaskKey, DirectoryStreamStateHandle.of(managedPath));
         }
     }
 

diff --git a/...rc/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java b/...rc/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
@@ -73,7 +73,6 @@
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
-import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -675,17 +674,13 @@ OperatorStateHandle deserializeOperatorStateHandle(
                 Preconditions.checkArgument(stateHandle instanceof SegmentFileStateHandle);
                 return isEmpty
                         ? new EmptyFileMergingOperatorStreamStateHandle(
-                                DirectoryStreamStateHandle.forPathWithZeroSize(
-                                        new File(taskOwnedDirPathStr).toPath()),
-                                DirectoryStreamStateHandle.forPathWithZeroSize(
-                                        new File(sharedDirPathStr).toPath()),
+                                DirectoryStreamStateHandle.of(new Path(taskOwnedDirPathStr)),
+                                DirectoryStreamStateHandle.of(new Path(sharedDirPathStr)),
                                 offsetsMap,
                                 stateHandle)
                         : new FileMergingOperatorStreamStateHandle(
-                                DirectoryStreamStateHandle.forPathWithZeroSize(
-                                        new File(taskOwnedDirPathStr).toPath()),
-                                DirectoryStreamStateHandle.forPathWithZeroSize(
-                                        new File(sharedDirPathStr).toPath()),
+                                DirectoryStreamStateHandle.of(new Path(taskOwnedDirPathStr)),
+                                DirectoryStreamStateHandle.of(new Path(sharedDirPathStr)),
                                 offsetsMap,
                                 stateHandle);
             } else {

diff --git a/.../src/main/java/org/apache/flink/runtime/state/filemerging/DirectoryStreamStateHandle.java b/.../src/main/java/org/apache/flink/runtime/state/filemerging/DirectoryStreamStateHandle.java
@@ -19,23 +19,28 @@
 package org.apache.flink.runtime.state.filemerging;
 
 import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.runtime.state.DirectoryStateHandle;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
 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 {
+/**
+ * This state handle represents a directory, usually used to be registered to {@link
+ * org.apache.flink.runtime.state.SharedStateRegistry} to track the life cycle of the directory.
+ */
+public class DirectoryStreamStateHandle implements StreamStateHandle {
 
     private static final long serialVersionUID = 1L;
 
-    public DirectoryStreamStateHandle(@Nonnull Path directory, long directorySize) {
-        super(directory, directorySize);
+    private final Path directory;
+
+    public DirectoryStreamStateHandle(@Nonnull Path directory) {
+        this.directory = directory;
     }
 
     @Override
@@ -50,11 +55,16 @@ public Optional<byte[]> asBytesIfInMemory() {
 
     @Override
     public PhysicalStateHandleID getStreamStateHandleID() {
-        return new PhysicalStateHandleID(getDirectory().toString());
+        return new PhysicalStateHandleID(directory.toString());
     }
 
     public SharedStateRegistryKey createStateRegistryKey() {
-        return new SharedStateRegistryKey(getDirectory().toUri().toString());
+        return new SharedStateRegistryKey(directory.toString());
+    }
+
+    @Override
+    public int hashCode() {
+        return directory.hashCode();
     }
 
     @Override
@@ -68,23 +78,34 @@ public boolean equals(Object o) {
 
         DirectoryStreamStateHandle that = (DirectoryStreamStateHandle) o;
 
-        return getDirectory().equals(that.getDirectory());
+        return directory.equals(that.directory);
     }
 
     @Override
     public String toString() {
         return "DirectoryStreamStateHandle{" + "directory=" + getDirectory() + '}';
     }
 
+    public Path getDirectory() {
+        return directory;
+    }
+
+    @Override
+    public void discardState() throws Exception {
+        FileSystem fs = directory.getFileSystem();
+        fs.delete(directory, true);
+    }
+
     /**
-     * Return a {@link DirectoryStreamStateHandle} with zero size, which usually used to be
-     * registered to {@link org.apache.flink.runtime.state.SharedStateRegistry} to track the life
-     * cycle of the directory, therefore a fake size is provided.
-     *
-     * @param directory the directory.
-     * @return DirectoryStreamStateHandle with zero size.
+     * This handle usually used to track the life cycle of the directory, therefore a fake size is
+     * provided.
      */
-    public static DirectoryStreamStateHandle forPathWithZeroSize(@Nonnull Path directory) {
-        return new DirectoryStreamStateHandle(directory, 0);
+    @Override
+    public long getStateSize() {
+        return 0;
+    }
+
+    public static DirectoryStreamStateHandle of(@Nonnull Path directory) {
+        return new DirectoryStreamStateHandle(directory);
     }
 }
diff --git a/...ntime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/CheckpointTestUtils.java b/...ntime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/CheckpointTestUtils.java
@@ -50,7 +50,6 @@
 
 import javax.annotation.Nullable;
 
-import java.io.File;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -212,11 +211,9 @@ private static OperatorStreamStateHandle createDummyOperatorStreamStateHandle(Ra
         boolean enableFileMerging = rnd.nextBoolean();
         if (enableFileMerging) {
             DirectoryStreamStateHandle taskOwnedDirHandle =
-                    DirectoryStreamStateHandle.forPathWithZeroSize(
-                            new File(String.valueOf(createRandomUUID(rnd))).toPath());
+                    DirectoryStreamStateHandle.of(new Path(createRandomUUID(rnd).toString()));
             DirectoryStreamStateHandle sharedDirHandle =
-                    DirectoryStreamStateHandle.forPathWithZeroSize(
-                            new File(String.valueOf(createRandomUUID(rnd))).toPath());
+                    DirectoryStreamStateHandle.of(new Path(createRandomUUID(rnd).toString()));
             return rnd.nextBoolean()
                     ? new FileMergingOperatorStreamStateHandle(
                             taskOwnedDirHandle,