diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java index 86d927adc640dfe46e3bba8e094a9509cf7f5ef8..6a495646d9dde4d04ce1ac587191fbba38454c9f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java @@ -68,11 +68,11 @@ public class JobClient { private static final Logger LOG = LoggerFactory.getLogger(JobClient.class); - public static ActorSystem startJobClientActorSystem(Configuration config) + public static ActorSystem startJobClientActorSystem(Configuration config, String hostname) throws IOException { LOG.info("Starting JobClient actor system"); - Option> remoting = new Some<>(new Tuple2("", 0)); + Option> remoting = new Some<>(new Tuple2(hostname, 0)); // start a remote actor system to listen on an arbitrary port ActorSystem system = AkkaUtils.createActorSystem(config, remoting); diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala index a82905936940e2035d9fe7e876d35a2b2be69d37..0e6f102fc6b14c50a923056748d230b57db7be19 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala @@ -310,7 +310,7 @@ abstract class FlinkMiniCluster( "The FlinkMiniCluster has not been started yet.") } } else { - JobClient.startJobClientActorSystem(originalConfiguration) + JobClient.startJobClientActorSystem(originalConfiguration, hostname) } }