[FLINK-35332][yarn] Do not set "rest.address" to the actual IP address of the YARN NM to support dual network envs

Co-authored-by: Paul Zhang <[email protected]>

Author: gkomlossi

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

@@ -174,7 +174,7 @@ rest:
   #
   # To enable this, set the bind address to one that has access to outside-facing
   # network interface, such as 0.0.0.0.
-  bind-address: localhost
+  # bind-address: localhost
   # # The port to which the REST client connects to. If rest.bind-port has
   # # not been specified, then the server will bind to this port as well.
   # port: 8081

flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java

@@ -133,7 +133,7 @@ void testGetJmResourceParamsWithDynamicProperties() throws Exception {
 
     @Test
     void testGetConfiguration() throws Exception {
-        int expectedResultLines = 26;
+        int expectedResultLines = 25;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),
@@ -209,7 +209,7 @@ void testMigrateLegacyConfigToStandardYaml() throws Exception {
 
     @Test
     void testGetConfigurationRemoveKey() throws Exception {
-        int expectedResultLines = 24;
+        int expectedResultLines = 23;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),
@@ -226,7 +226,7 @@ void testGetConfigurationRemoveKey() throws Exception {
 
     @Test
     void testGetConfigurationRemoveKeyValue() throws Exception {
-        int expectedResultLines = 24;
+        int expectedResultLines = 23;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),
@@ -243,7 +243,7 @@ void testGetConfigurationRemoveKeyValue() throws Exception {
 
     @Test
     void testGetConfigurationRemoveKeyValueNotMatchingValue() throws Exception {
-        int expectedResultLines = 26;
+        int expectedResultLines = 25;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),
@@ -260,7 +260,7 @@ void testGetConfigurationRemoveKeyValueNotMatchingValue() throws Exception {
 
     @Test
     void testGetConfigurationReplaceKeyValue() throws Exception {
-        int expectedResultLines = 26;
+        int expectedResultLines = 25;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),
@@ -277,7 +277,7 @@ void testGetConfigurationReplaceKeyValue() throws Exception {
 
     @Test
     void testGetConfigurationReplaceKeyValueNotMatchingValue() throws Exception {
-        int expectedResultLines = 26;
+        int expectedResultLines = 25;
         String[] commands = {
             RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
             BashJavaUtils.Command.UPDATE_AND_GET_FLINK_CONFIGURATION.toString(),

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

@@ -65,7 +65,9 @@ public static Configuration loadConfiguration(
 
         configuration.set(JobManagerOptions.ADDRESS, hostname);
         configuration.set(RestOptions.ADDRESS, hostname);
-        configuration.set(RestOptions.BIND_ADDRESS, hostname);
+        if (!configuration.contains(RestOptions.BIND_ADDRESS)) {
+            configuration.set(RestOptions.BIND_ADDRESS, hostname);
+        }
 
         // if a web monitor shall be started, set the port to random binding
         if (configuration.get(WebOptions.PORT, 0) >= 0) {

flink-yarn/src/test/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtilsTest.java

@@ -89,6 +89,24 @@ void testRestPortAndBindingPortSpecified() throws IOException {
         assertThat(configuration.get(RestOptions.BIND_PORT)).isEqualTo(bindingPortRange);
     }
 
+    @Test
+    void testRestBindingAddressUnspecified() throws IOException {
+        final Configuration initialConfiguration = new Configuration();
+        final Configuration configuration = loadConfiguration(initialConfiguration);
+
+        assertThat(configuration.get(RestOptions.BIND_ADDRESS)).isEqualTo("foobar");
+    }
+
+    @Test
+    void testRestBindingAddressSpecified() throws Exception {
+        final Configuration initialConfiguration = new Configuration();
+        final String bindingAddress = "0.0.0.0";
+        initialConfiguration.set(RestOptions.BIND_ADDRESS, bindingAddress);
+        final Configuration configuration = loadConfiguration(initialConfiguration);
+
+        assertThat(configuration.get(RestOptions.BIND_ADDRESS)).isEqualTo(bindingAddress);
+    }
+
     @Test
     void testParsingValidKerberosEnv() throws IOException {
         final Configuration initialConfiguration = new Configuration();