[FLINK-24367] RpcSystem#loader evaluates all service entries

nicusX · Oct 4, 2021 · cf272b2 · cf272b2
1 parent 458019b
commit cf272b2
Showing 1 changed file with 15 additions and 1 deletion.
diff --git a/flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcSystem.java b/flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcSystem.java
@@ -18,9 +18,11 @@
 package org.apache.flink.runtime.rpc;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.ExceptionUtils;
 
 import javax.annotation.Nullable;
 
+import java.util.Iterator;
 import java.util.ServiceLoader;
 
 /**
@@ -88,7 +90,19 @@ static RpcSystem load() {
      * @return loaded RpcSystem
      */
     static RpcSystem load(Configuration config) {
-        return ServiceLoader.load(RpcSystemLoader.class).iterator().next().loadRpcSystem(config);
+        final Iterator<RpcSystemLoader> iterator =
+                ServiceLoader.load(RpcSystemLoader.class).iterator();
+
+        Exception loadError = null;
+        while (iterator.hasNext()) {
+            final RpcSystemLoader next = iterator.next();
+            try {
+                return next.loadRpcSystem(config);
+            } catch (Exception e) {
+                loadError = ExceptionUtils.firstOrSuppressed(e, loadError);
+            }
+        }
+        throw new RuntimeException("Could not load RpcSystem.", loadError);
     }
 
     /** Descriptor for creating a fork-join thread-pool. */