[SPARK-45278][YARN] Allow limited configuration option to either HOSTNAME or ALL_IPS

Author: hendra.saputra

docs/running-on-yarn.md

@@ -515,12 +515,10 @@ To use a custom metrics.properties for the application master and executors, upd
   <td>1.6.0</td>
 </tr>
 <tr>
-  <td><code>spark.yarn.executor.bindAddress</code></td>
-  <td><code>(executor hostname)</code></td>
+  <td><code>spark.yarn.executor.bindAddress.mode</code></td>
+  <td><code><code>HOSTNAME</code></code></td>
   <td>
-  Hostname or IP address where to bind listening sockets in YARN cluster and client mode.
-  <br />It also allows a different address from the local one to be advertised to other
-  executors or external systems.
+  Configures executor behaviour of which network to listen sockets to. Possible choices are: <code>HOSTNAME</code> means to bind to hostname, <code>ALL_IPS</code> means to bind to 0.0.0.0
   </td>
   <td>4.0.0</td>
 </tr>

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -176,7 +176,7 @@ private[yarn] class YarnAllocator(
 
   private val minMemoryOverhead = sparkConf.get(EXECUTOR_MIN_MEMORY_OVERHEAD)
 
-  private val bindAddress = sparkConf.get(EXECUTOR_BIND_ADDRESS)
+  private val bindAddressMode = sparkConf.get(EXECUTOR_BIND_ADDRESS_MODE)
 
   private val memoryOverheadFactor = sparkConf.get(EXECUTOR_MEMORY_OVERHEAD_FACTOR)
 
@@ -762,7 +762,10 @@ private[yarn] class YarnAllocator(
       val rpId = getResourceProfileIdFromPriority(container.getPriority)
       executorIdCounter += 1
       val executorHostname = container.getNodeId.getHost
-      val executorBindAddress = bindAddress.getOrElse(executorHostname)
+      val executorBindAddress = bindAddressMode match {
+        case "ALL_IPS" => "0.0.0.0"
+        case _ => executorHostname
+      }
       val containerId = container.getId
       val executorId = executorIdCounter.toString
       val yarnResourceForRpId = rpIdToYarnResource.get(rpId)
@@ -846,7 +849,10 @@ private[yarn] class YarnAllocator(
         new HashSet[ContainerId])
       containerSet += containerId
       allocatedContainerToHostMap.put(containerId, executorHostname)
-      allocatedContainerToBindAddressMap.put(containerId, bindAddress.getOrElse(executorHostname))
+      allocatedContainerToBindAddressMap.put(containerId, bindAddressMode match {
+        case "ALL_IPS" => "0.0.0.0"
+        case _ => executorHostname
+      })
       launchingExecutorContainerIds.remove(containerId)
     }
     getOrUpdateNumExecutorsStartingForRPId(rpId).decrementAndGet()

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy.yarn
 
-import java.util.Properties
+import java.util.{Locale, Properties}
 import java.util.concurrent.TimeUnit
 
 import org.apache.spark.internal.Logging
@@ -339,12 +339,16 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
-  private[spark] val EXECUTOR_BIND_ADDRESS =
-    ConfigBuilder("spark.yarn.executor.bindAddress")
-      .doc("Address where to bind network listen sockets on the executor.")
+  private[spark] val EXECUTOR_BIND_ADDRESS_MODE =
+    ConfigBuilder("spark.yarn.executor.bindAddress.mode")
+      .doc("Configures executor behaviour of which network to listen sockets to. " +
+        "Possible choices are: HOSTNAME means to bind to hostname, " +
+        "ALL_IPS means to bind to 0.0.0.0")
       .version("4.0.0")
       .stringConf
-      .createOptional
+      .transform(_.toUpperCase(Locale.ROOT))
+      .checkValues(Set("HOSTNAME", "ALL_IPS"))
+      .createWithDefault("HOSTNAME")
 
   /* Unmanaged AM configuration. */
 

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala

@@ -905,9 +905,9 @@ class YarnAllocatorSuite extends SparkFunSuite
     handler.getNumExecutorsStarting should be(0)
   }
 
-  test("use requested bind-address") {
+  test("use 0.0.0.0 when requested bind-address mode is ALL_IPS") {
     val (handler, _) = createAllocator(maxExecutors = 1,
-        additionalConfigs = Map(EXECUTOR_BIND_ADDRESS.key -> "0.0.0.0"))
+      additionalConfigs = Map(EXECUTOR_BIND_ADDRESS_MODE.key -> "ALL_IPS"))
     handler.updateResourceRequests()
 
     val container = createContainer("host1")
@@ -917,4 +917,16 @@ class YarnAllocatorSuite extends SparkFunSuite
     handler.allocatedContainerToBindAddressMap.get(container.getId).get should be ("0.0.0.0")
   }
 
+  test("use hostname when requested bind-address mode is HOSTNAME") {
+    val (handler, _) = createAllocator(maxExecutors = 1,
+      additionalConfigs = Map(EXECUTOR_BIND_ADDRESS_MODE.key -> "HOSTNAME"))
+    handler.updateResourceRequests()
+
+    val container = createContainer("host1")
+    handler.handleAllocatedContainers(Array(container).toImmutableArraySeq)
+
+    handler.allocatedContainerToHostMap.get(container.getId).get should be ("host1")
+    handler.allocatedContainerToBindAddressMap.get(container.getId).get should be ("host1")
+  }
+
 }