[BEAM-9401] bind Flink MiniCluster to localhost

grtheod · Feb 27, 2020 · 5db8660 · 5db8660
1 parent b9ca34d
commit 5db8660
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/runners/flink/job-server/test_flink_uber_jar.sh b/runners/flink/job-server/test_flink_uber_jar.sh
@@ -91,7 +91,7 @@ s.close()
 FLINK_PORT=$(python -c "$SOCKET_SCRIPT")
 
 echo "Starting Flink mini cluster listening on port $FLINK_PORT"
-java -Dorg.slf4j.simpleLogger.defaultLogLevel=warn -jar "$FLINK_MINI_CLUSTER_JAR" --rest-port "$FLINK_PORT" &
+java -Dorg.slf4j.simpleLogger.defaultLogLevel=warn -jar "$FLINK_MINI_CLUSTER_JAR" --rest-port "$FLINK_PORT" --rest-bind-address localhost &
 
 PIPELINE_PY="
 import apache_beam as beam

diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkMiniClusterEntryPoint.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkMiniClusterEntryPoint.java
@@ -36,6 +36,9 @@ static class MiniClusterArgs {
     @Option(name = "--rest-port")
     int restPort = 0;
 
+    @Option(name = "--rest-bind-address")
+    String restBindAddress = "";
+
     @Option(name = "--num-task-managers")
     int numTaskManagers = 1;
 
@@ -48,6 +51,9 @@ public static void main(String[] args) throws Exception {
 
     Configuration flinkConfig = new Configuration();
     flinkConfig.setInteger(RestOptions.PORT, miniClusterArgs.restPort);
+    if (!miniClusterArgs.restBindAddress.isEmpty()) {
+      flinkConfig.setString(RestOptions.BIND_ADDRESS, miniClusterArgs.restBindAddress);
+    }
 
     MiniClusterConfiguration clusterConfig =
         new MiniClusterConfiguration.Builder()