[FLINK-38433][state/forst] Avoid delete ForSt's directory when there happened to be an existing one (#27064)

(cherry picked from commit a644a82)

Author: AlexYinHan

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

@@ -86,6 +86,8 @@ public final class ForStResourceContainer implements AutoCloseable {
 
     @Nullable private final Path remoteForStPath;
 
+    private boolean remotePathNewlyCreated;
+
     @Nullable private final Path localBasePath;
 
     @Nullable private final Path localForStPath;
@@ -369,7 +371,7 @@ public int getWriteIoParallelism() {
      */
     public void prepareDirectories() throws Exception {
         if (remoteBasePath != null && remoteForStPath != null) {
-            prepareDirectories(remoteBasePath, remoteForStPath);
+            remotePathNewlyCreated = prepareDirectories(remoteBasePath, remoteForStPath);
         }
         if (localBasePath != null && localForStPath != null) {
             prepareDirectories(
@@ -402,23 +404,26 @@ public void prepareDirectories() throws Exception {
         return forStFileSystem;
     }
 
-    private static void prepareDirectories(Path basePath, Path dbPath) throws IOException {
+    private static boolean prepareDirectories(Path basePath, Path dbPath) throws IOException {
+        boolean allNewlyCreated = true;
         FileSystem fileSystem = basePath.getFileSystem();
         if (fileSystem.exists(basePath)) {
             if (!fileSystem.getFileStatus(basePath).isDir()) {
                 throw new IOException("Not a directory: " + basePath);
             }
+            allNewlyCreated = false;
         } else if (!fileSystem.mkdirs(basePath)) {
             throw new IOException(
                     String.format("Could not create ForSt directory at %s.", basePath));
         }
         if (fileSystem.exists(dbPath)) {
-            fileSystem.delete(dbPath, true);
-        }
-        if (!fileSystem.mkdirs(dbPath)) {
+            LOG.info("Reusing previous ForSt db directory at {}.", dbPath);
+            allNewlyCreated = false;
+        } else if (!fileSystem.mkdirs(dbPath)) {
             throw new IOException(
                     String.format("Could not create ForSt db directory at %s.", dbPath));
         }
+        return allNewlyCreated;
     }
 
     /**
@@ -436,7 +441,7 @@ public void clearDirectories() throws Exception {
     }
 
     public void forceClearRemoteDirectories() throws Exception {
-        if (remoteBasePath != null) {
+        if (remoteBasePath != null && remotePathNewlyCreated) {
             clearDirectories(remoteBasePath);
         }
     }

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java

@@ -336,7 +336,9 @@ public void testDirectoryResources() throws Exception {
 
             assertTrue(new File(remoteBasePath.getPath()).exists());
             optionsContainer.forceClearRemoteDirectories();
-            assertFalse(new File(remoteBasePath.getPath()).exists());
+
+            // Do not delete remote directory because it is not created by ForStResourceContainer
+            assertTrue(new File(remoteBasePath.getPath()).exists());
         }
     }