[FLINK-27118][yarn] TM ignores localhost BIND_HOST

This closes apache#19395.

flink-dist/src/main/resources/flink-conf.yaml

@@ -38,6 +38,8 @@ jobmanager.rpc.port: 6123
 
 # The host interface the JobManager will bind to. My default, this is localhost, and will prevent
 # the JobManager from communicating outside the machine/container it is running on.
+# On YARN this setting will be ignored if it is set to 'localhost', defaulting to 0.0.0.0.
+# On Kubernetes this setting will be ignored, defaulting to 0.0.0.0.
 #
 # To enable this, set the bind-host address to one that has access to an outside facing network
 # interface, such as 0.0.0.0.
@@ -53,6 +55,8 @@ jobmanager.memory.process.size: 1600m
 
 # The host interface the TaskManager will bind to. By default, this is localhost, and will prevent
 # the TaskManager from communicating outside the machine/container it is running on.
+# On YARN this setting will be ignored if it is set to 'localhost', defaulting to 0.0.0.0.
+# On Kubernetes this setting will be ignored, defaulting to 0.0.0.0.
 #
 # To enable this, set the bind-host address to one that has access to an outside facing network
 # interface, such as 0.0.0.0.

flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java

@@ -42,6 +42,7 @@
 import org.apache.flink.configuration.ResourceManagerOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.plugin.PluginConfig;
 import org.apache.flink.core.plugin.PluginUtils;
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
@@ -1020,6 +1021,13 @@ private ApplicationReport startAppMaster(
  File tmpConfigurationFile = null;
  try {
  tmpConfigurationFile = File.createTempFile(appId + "-flink-conf.yaml", null);
+
+ // remove localhost bind hosts as they render production clusters unusable
+ removeLocalhostBindHostSetting(configuration, JobManagerOptions.BIND_HOST);
+ removeLocalhostBindHostSetting(configuration, TaskManagerOptions.BIND_HOST);
+ // this setting is unconditionally overridden anyway, so we remove it for clarity
+ configuration.removeConfig(TaskManagerOptions.HOST);
+
  BootstrapTools.writeConfiguration(configuration, tmpConfigurationFile);
 
  String flinkConfigKey = "flink-conf.yaml";
@@ -1266,6 +1274,20 @@ private ApplicationReport startAppMaster(
  return report;
  }
 
+ private void removeLocalhostBindHostSetting(
+ Configuration configuration, ConfigOption<?> option) {
+ configuration
+ .getOptional(option)
+ .filter(bindHost -> bindHost.equals("localhost"))
+ .ifPresent(
+ bindHost -> {
+ LOG.info(
+ "Removing 'localhost' {} setting from effective configuration; using '0.0.0.0' instead.",
+ option);
+ configuration.removeConfig(option);
+ });
+ }
+
  private void setTokensFor(ContainerLaunchContext containerLaunchContext) throws IOException {
  LOG.info("Adding delegation tokens to the AM container.");
 

flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtils.java

@@ -25,7 +25,6 @@
 import org.apache.flink.configuration.ResourceManagerOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.SecurityOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.configuration.WebOptions;
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
 import org.apache.flink.util.Preconditions;
@@ -65,9 +64,6 @@ public static Configuration loadConfiguration(
  ApplicationConstants.Environment.NM_HOST.key());
 
  configuration.setString(JobManagerOptions.ADDRESS, hostname);
- configuration.removeConfig(JobManagerOptions.BIND_HOST);
- configuration.removeConfig(TaskManagerOptions.BIND_HOST);
- configuration.removeConfig(TaskManagerOptions.HOST);
  configuration.setString(RestOptions.ADDRESS, hostname);
  configuration.setString(RestOptions.BIND_ADDRESS, hostname);