[FLINK-35410][state] Avoid sync waiting in coordinator thread of ForSt executor (apache#24819)

Author: Zakelly

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

@@ -21,7 +21,6 @@
 import org.apache.flink.runtime.asyncprocessing.StateExecutor;
 import org.apache.flink.runtime.asyncprocessing.StateRequest;
 import org.apache.flink.runtime.asyncprocessing.StateRequestContainer;
-import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 import org.apache.flink.util.concurrent.FutureUtils;
@@ -34,7 +33,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -77,7 +75,8 @@ public CompletableFuture<Void> executeBatchRequests(
         Preconditions.checkArgument(stateRequestContainer instanceof ForStStateRequestClassifier);
         ForStStateRequestClassifier stateRequestClassifier =
                 (ForStStateRequestClassifier) stateRequestContainer;
-        return CompletableFuture.supplyAsync(
+        CompletableFuture<Void> resultFuture = new CompletableFuture<>();
+        coordinatorThread.execute(
                 () -> {
                     long startTime = System.currentTimeMillis();
                     List<CompletableFuture<Void>> futures = new ArrayList<>(2);
@@ -98,22 +97,20 @@ public CompletableFuture<Void> executeBatchRequests(
                         futures.add(getOperations.process());
                     }
 
-                    try {
-                        FutureUtils.waitForAll(futures).get();
-                    } catch (InterruptedException | ExecutionException e) {
-                        throw new FlinkRuntimeException(
-                                "Exception when executing ForSt writeBatch or multiGet operation",
-                                e);
-                    }
-                    long duration = System.currentTimeMillis() - startTime;
-                    LOG.debug(
-                            "Complete executing a batch of state requests, putRequest size {}, getRequest size {}, duration {} ms",
-                            putRequests.size(),
-                            getRequests.size(),
-                            duration);
-                    return null;
-                },
-                coordinatorThread);
+                    FutureUtils.combineAll(futures)
+                            .thenAcceptAsync(
+                                    (e) -> {
+                                        long duration = System.currentTimeMillis() - startTime;
+                                        LOG.debug(
+                                                "Complete executing a batch of state requests, putRequest size {}, getRequest size {}, duration {} ms",
+                                                putRequests.size(),
+                                                getRequests.size(),
+                                                duration);
+                                        resultFuture.complete(null);
+                                    },
+                                    coordinatorThread);
+                });
+        return resultFuture;
     }
 
     @Override

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

@@ -57,7 +57,7 @@ public void testExecuteValueStateRequest() throws Exception {
                     buildStateRequest(state, StateRequestType.VALUE_UPDATE, i, "test-" + i, i * 2));
         }
 
-        forStStateExecutor.executeBatchRequests(stateRequestContainer);
+        forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
 
         List<StateRequest<?, ?, ?>> checkList = new ArrayList<>();
         stateRequestContainer = forStStateExecutor.createStateRequestContainer();
@@ -99,7 +99,7 @@ public void testExecuteValueStateRequest() throws Exception {
             stateRequestContainer.offer(
                     buildStateRequest(state, StateRequestType.VALUE_UPDATE, i, null, i * 2));
         }
-        forStStateExecutor.executeBatchRequests(stateRequestContainer);
+        forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
 
         // 5. Check that the deleted value is null :  keyRange [keyNum - 100, keyNum + 100)
         stateRequestContainer = forStStateExecutor.createStateRequestContainer();
@@ -111,7 +111,7 @@ public void testExecuteValueStateRequest() throws Exception {
             stateRequestContainer.offer(getRequest);
             checkList.add(getRequest);
         }
-        forStStateExecutor.executeBatchRequests(stateRequestContainer);
+        forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
         for (StateRequest<?, ?, ?> getRequest : checkList) {
             assertThat(getRequest.getRequestType()).isEqualTo(StateRequestType.VALUE_GET);
             assertThat(((TestStateFuture<String>) getRequest.getFuture()).getCompletedResult())