[FLINK-34975][state/forst] Execute read/write state request in different executor (apache#25360)

Author: fredia

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

@@ -39,6 +39,8 @@
 import static org.apache.flink.configuration.ConfigOptions.key;
 import static org.apache.flink.configuration.description.LinkElement.link;
 import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.state.forst.ForStOptions.EXECUTOR_READ_IO_PARALLELISM;
+import static org.apache.flink.state.forst.ForStOptions.EXECUTOR_WRITE_IO_PARALLELISM;
 import static org.rocksdb.CompactionStyle.FIFO;
 import static org.rocksdb.CompactionStyle.LEVEL;
 import static org.rocksdb.CompactionStyle.NONE;
@@ -322,6 +324,9 @@ public class ForStConfigurableOptions implements Serializable {
 
     static final ConfigOption<?>[] CANDIDATE_CONFIGS =
             new ConfigOption<?>[] {
+                // configurable forst executor
+                EXECUTOR_WRITE_IO_PARALLELISM,
+                EXECUTOR_READ_IO_PARALLELISM,
                 // configurable DBOptions
                 MAX_BACKGROUND_THREADS,
                 MAX_OPEN_FILES,

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

@@ -274,9 +274,12 @@ public StateExecutor createStateExecutor() {
                 throw new FlinkRuntimeException(
                         "Attempt to create StateExecutor after ForStKeyedStateBackend is disposed.");
             }
-            // TODO: Make io parallelism configurable
             StateExecutor stateExecutor =
-                    new ForStStateExecutor(4, db, optionsContainer.getWriteOptions());
+                    new ForStStateExecutor(
+                            optionsContainer.getReadIoParallelism(),
+                            optionsContainer.getWriteIoParallelism(),
+                            db,
+                            optionsContainer.getWriteOptions());
             managedStateExecutors.add(stateExecutor);
             return stateExecutor;
         }

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

@@ -156,4 +156,18 @@ public class ForStOptions {
                                             + "when '%s' is configured to '%s'. Increasing this value can improve the performance "
                                             + "of rocksdb timer service, but consumes more heap memory at the same time.",
                                     TIMER_SERVICE_FACTORY.key(), ForStDB.name()));
+
+    public static final ConfigOption<Integer> EXECUTOR_READ_IO_PARALLELISM =
+            ConfigOptions.key("state.backend.forst.memory.executor-read-io-parallelism")
+                    .intType()
+                    .defaultValue(3)
+                    .withDescription(
+                            "The number of threads used for read IO operations in the executor.");
+
+    public static final ConfigOption<Integer> EXECUTOR_WRITE_IO_PARALLELISM =
+            ConfigOptions.key("state.backend.forst.memory.executor-write-io-parallelism")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "The number of threads used for write IO operations in the executor.");
 }

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

@@ -277,6 +277,14 @@ public Path getDbPath() {
         }
     }
 
+    public int getReadIoParallelism() {
+        return configuration.get(ForStOptions.EXECUTOR_READ_IO_PARALLELISM);
+    }
+
+    public int getWriteIoParallelism() {
+        return configuration.get(ForStOptions.EXECUTOR_WRITE_IO_PARALLELISM);
+    }
+
     /**
      * Prepare local and remote directories.
      *

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

@@ -52,22 +52,41 @@ public class ForStStateExecutor implements StateExecutor {
      */
     private final ExecutorService coordinatorThread;
 
-    /** The worker thread that actually executes the {@link StateRequest}s. */
-    private final ExecutorService workerThreads;
+    /** The worker thread that actually executes the read {@link StateRequest}s. */
+    private final ExecutorService readThreads;
+
+    /** The worker thread that actually executes the write {@link StateRequest}s. */
+    private final ExecutorService writeThreads;
 
     private final RocksDB db;
 
     private final WriteOptions writeOptions;
 
     private Throwable executionError;
 
-    public ForStStateExecutor(int ioParallelism, RocksDB db, WriteOptions writeOptions) {
+    public ForStStateExecutor(
+            int readIoParallelism, int writeIoParallelism, RocksDB db, WriteOptions writeOptions) {
+        Preconditions.checkState(readIoParallelism > 0 || writeIoParallelism > 0);
         this.coordinatorThread =
                 Executors.newSingleThreadScheduledExecutor(
                         new ExecutorThreadFactory("ForSt-StateExecutor-Coordinator"));
-        this.workerThreads =
-                Executors.newFixedThreadPool(
-                        ioParallelism, new ExecutorThreadFactory("ForSt-StateExecutor-IO"));
+        if (readIoParallelism <= 0 || writeIoParallelism <= 0) {
+            this.readThreads =
+                    Executors.newFixedThreadPool(
+                            Math.max(readIoParallelism, writeIoParallelism),
+                            new ExecutorThreadFactory("ForSt-StateExecutor-IO"));
+            this.writeThreads = null;
+        } else {
+            this.readThreads =
+                    Executors.newFixedThreadPool(
+                            readIoParallelism,
+                            new ExecutorThreadFactory("ForSt-StateExecutor-read-IO"));
+
+            this.writeThreads =
+                    Executors.newFixedThreadPool(
+                            writeIoParallelism,
+                            new ExecutorThreadFactory("ForSt-StateExecutor-write-IO"));
+        }
         this.db = db;
         this.writeOptions = writeOptions;
     }
@@ -89,23 +108,26 @@ public CompletableFuture<Void> executeBatchRequests(
                     if (!putRequests.isEmpty()) {
                         ForStWriteBatchOperation writeOperations =
                                 new ForStWriteBatchOperation(
-                                        db, putRequests, writeOptions, workerThreads);
+                                        db,
+                                        putRequests,
+                                        writeOptions,
+                                        writeThreads == null ? readThreads : writeThreads);
                         futures.add(writeOperations.process());
                     }
 
                     List<ForStDBGetRequest<?, ?, ?, ?>> getRequests =
                             stateRequestClassifier.pollDbGetRequests();
                     if (!getRequests.isEmpty()) {
                         ForStGeneralMultiGetOperation getOperations =
-                                new ForStGeneralMultiGetOperation(db, getRequests, workerThreads);
+                                new ForStGeneralMultiGetOperation(db, getRequests, readThreads);
                         futures.add(getOperations.process());
                     }
 
                     List<ForStDBIterRequest<?, ?, ?, ?, ?>> iterRequests =
                             stateRequestClassifier.pollDbIterRequests();
                     if (!iterRequests.isEmpty()) {
                         ForStIterateOperation iterOperations =
-                                new ForStIterateOperation(db, iterRequests, workerThreads);
+                                new ForStIterateOperation(db, iterRequests, readThreads);
                         futures.add(iterOperations.process());
                     }
 
@@ -155,7 +177,10 @@ private void checkState() {
 
     @Override
     public void shutdown() {
-        workerThreads.shutdown();
+        readThreads.shutdown();
+        if (writeThreads != null) {
+            writeThreads.shutdown();
+        }
         coordinatorThread.shutdown();
         LOG.info("Shutting down the ForStStateExecutor.");
     }

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

@@ -46,7 +46,8 @@ class ForStStateExecutorTest extends ForStDBOperationTestBase {
     @Test
     @SuppressWarnings("unchecked")
     void testExecuteValueStateRequest() throws Exception {
-        ForStStateExecutor forStStateExecutor = new ForStStateExecutor(4, db, new WriteOptions());
+        ForStStateExecutor forStStateExecutor =
+                new ForStStateExecutor(3, 1, db, new WriteOptions());
         ForStValueState<Integer, VoidNamespace, String> state1 =
                 buildForStValueState("value-state-1");
         ForStValueState<Integer, VoidNamespace, String> state2 =
@@ -129,7 +130,8 @@ void testExecuteValueStateRequest() throws Exception {
 
     @Test
     void testExecuteMapStateRequest() throws Exception {
-        ForStStateExecutor forStStateExecutor = new ForStStateExecutor(4, db, new WriteOptions());
+        ForStStateExecutor forStStateExecutor =
+                new ForStStateExecutor(3, 1, db, new WriteOptions());
         ForStMapState<Integer, VoidNamespace, String, String> state =
                 buildForStMapState("map-state");
         StateRequestContainer stateRequestContainer =