[FLINK-36395][state/forst] Graceful quit for forst threads

Zakelly · Sep 27, 2024 · 8582cce · 8582cce
1 parent 82582b3
commit 8582cce
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 20 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AsyncKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AsyncKeyedStateBackend.java
@@ -31,8 +31,6 @@
 
 import javax.annotation.Nonnull;
 
-import java.io.Closeable;
-
 /**
  * An async keyed state backend provides methods supporting to access keyed state asynchronously and
  * in batch.
@@ -42,7 +40,6 @@ public interface AsyncKeyedStateBackend<K>
         extends Snapshotable<SnapshotResult<KeyedStateHandle>>,
                 InternalCheckpointListener,
                 Disposable,
-                Closeable,
                 AsyncExecutionController.SwitchContextListener<K> {
 
     /**

diff --git a/...statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java b/...statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
@@ -54,7 +54,6 @@
 import javax.annotation.concurrent.GuardedBy;
 
 import java.io.File;
-import java.io.IOException;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Set;
@@ -330,7 +329,9 @@ public void dispose() {
         }
         synchronized (lock) {
             if (!closed) {
-                IOUtils.closeQuietly(this);
+                for (StateExecutor executor : managedStateExecutors) {
+                    executor.shutdown();
+                }
             }
         }
 
@@ -382,19 +383,6 @@ Path getRemoteBasePath() {
         return optionsContainer.getRemoteBasePath();
     }
 
-    @Override
-    public void close() throws IOException {
-        synchronized (lock) {
-            if (closed) {
-                return;
-            }
-            closed = true;
-            for (StateExecutor executor : managedStateExecutors) {
-                executor.shutdown();
-            }
-        }
-    }
-
     /** ForSt specific information about the k/v states. */
     public static class ForStKvStateInfo implements AutoCloseable {
         public final ColumnFamilyHandle columnFamilyHandle;

diff --git a/...ink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java b/...ink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java
@@ -230,11 +230,12 @@ private void checkState() {
 
     @Override
     public void shutdown() {
+        // Coordinator should be shutdown before others, since it submit jobs to others.
+        coordinatorThread.shutdown();
         readThreads.shutdown();
         if (!sharedWriteThread) {
             writeThreads.shutdown();
         }
-        coordinatorThread.shutdown();
         LOG.info("Shutting down the ForStStateExecutor.");
     }
 }
diff --git a/...tatebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java b/...tatebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
@@ -49,8 +49,8 @@ public class ForStFlinkFileSystem extends FileSystem {
 
     private static final Map<String, String> remoteLocalMapping = new ConcurrentHashMap<>();
     private static final Function<String, Boolean> miscFileFilter = s -> !s.endsWith(".sst");
-    private static final FileSystem localFS = FileSystem.getLocalFileSystem();
 
+    private final FileSystem localFS = FileSystem.getLocalFileSystem();
     private final FileSystem delegateFS;
     private final String remoteBase;
     private final Function<String, Boolean> localFileFilter;