Revert "[hotfix][core] Generalizes FutureUtils#runAsync"

This reverts commit e64b2fd.
XComp · Nov 27, 2024 · d9d09b6 · d9d09b6
1 parent ff60c9d
commit d9d09b6
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 9 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java b/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java
@@ -24,7 +24,6 @@
 import org.apache.flink.util.FatalExitExceptionHandler;
 import org.apache.flink.util.function.RunnableWithException;
 import org.apache.flink.util.function.SupplierWithException;
-import org.apache.flink.util.function.ThrowingRunnable;
 
 import javax.annotation.Nullable;
 
@@ -923,14 +922,14 @@ public static <T> CompletableFuture<T> supplyAsync(
     }
 
     /**
-     * Returns a future which is completed when {@code runnable} is finished.
+     * Returns a future which is completed when {@link RunnableWithException} is finished.
      *
      * @param runnable represents the task
      * @param executor to execute the runnable
      * @return Future which is completed when runnable is finished
      */
     public static CompletableFuture<Void> runAsync(
-            ThrowingRunnable<? extends Throwable> runnable, Executor executor) {
+            RunnableWithException runnable, Executor executor) {
         return CompletableFuture.runAsync(
                 () -> {
                     try {

diff --git a/.../java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java b/.../java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java
@@ -484,9 +484,13 @@ public void confirmLeadership(UUID leaderSessionID, String leaderAddress) {
         @Override
         public CompletableFuture<Void> runAsLeader(
                 UUID leaderSessionId, ThrowingRunnable<? extends Throwable> callback) {
-            return FutureUtils.runAsync(
-                    () -> EmbeddedLeaderService.this.runAsLeader(this, leaderSessionId, callback),
-                    Executors.directExecutor());
+            try {
+                EmbeddedLeaderService.this.runAsLeader(this, leaderSessionId, callback);
+            } catch (LeadershipLostException e) {
+                return FutureUtils.completedExceptionally(e);
+            }
+
+            return FutureUtils.completedVoidFuture();
         }
 
         void shutdown(Exception cause) {

diff --git a/...e/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java b/...e/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java
@@ -36,6 +36,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -367,14 +368,18 @@ CompletableFuture<Void> runAsLeader(
             String componentId,
             UUID leaderSessionId,
             ThrowingRunnable<? extends Throwable> callback) {
-        return FutureUtils.runAsync(
+        return CompletableFuture.runAsync(
                 () -> {
                     synchronized (lock) {
                         if (!hasLeadership(componentId, leaderSessionId)) {
-                            throw new LeadershipLostException(leaderSessionId);
+                            throw new CompletionException(new LeadershipLostException(leaderSessionId));
                         }
 
-                        callback.run();
+                        try {
+                            callback.run();
+                        } catch (Throwable e) {
+                            throw new CompletionException(e);
+                        }
                     }
                 },
                 leadershipOperationExecutor);