Support executor bind address in Yarn executors

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -253,6 +253,11 @@ package object config {
   private[spark] val EXECUTOR_ID =
     ConfigBuilder("spark.executor.id").version("1.2.0").stringConf.createOptional
 
+  private[spark] val EXECUTOR_BIND_ADDRESS = ConfigBuilder("spark.executor.bindAddress")
+    .doc("Address where to bind network listen sockets on the executor.")
+    .stringConf
+    .createWithDefault(Utils.localHostName())
+
   private[spark] val EXECUTOR_CLASS_PATH =
     ConfigBuilder(SparkLauncher.EXECUTOR_EXTRA_CLASSPATH)
       .version("1.0.0")

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

@@ -461,7 +461,7 @@ private[spark] class ApplicationMaster(
       val executorMemory = _sparkConf.get(EXECUTOR_MEMORY).toInt
       val executorCores = _sparkConf.get(EXECUTOR_CORES)
       val dummyRunner = new ExecutorRunnable(None, yarnConf, _sparkConf, driverUrl, "<executorId>",
-        "<hostname>", executorMemory, executorCores, appId, securityMgr, localResources,
+        "<bindAddress>", "<hostname>", executorMemory, executorCores, appId, securityMgr, localResources,
         ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
       dummyRunner.launchContextDebugInfo()
     }

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

@@ -49,6 +49,7 @@ private[yarn] class ExecutorRunnable(
     sparkConf: SparkConf,
     masterAddress: String,
     executorId: String,
+    bindAddress: String,
     hostname: String,
     executorMemory: Int,
     executorCores: Int,
@@ -117,7 +118,7 @@ private[yarn] class ExecutorRunnable(
     } catch {
       case ex: Exception =>
         throw new SparkException(s"Exception while starting container ${container.get.getId}" +
-          s" on host $hostname", ex)
+          s" on host $hostname ($bindAddress)", ex)
     }
   }
 
@@ -189,6 +190,7 @@ private[yarn] class ExecutorRunnable(
       Seq("org.apache.spark.executor.YarnCoarseGrainedExecutorBackend",
         "--driver-url", masterAddress,
         "--executor-id", executorId,
+        "--bindAddress", bindAddress,
         "--hostname", hostname,
         "--cores", executorCores.toString,
         "--app-id", appId,

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

@@ -733,12 +733,13 @@ private[yarn] class YarnAllocator(
     for (container <- containersToUse) {
       val rpId = getResourceProfileIdFromPriority(container.getPriority)
       executorIdCounter += 1
+      val executorBindAddress = sparkConf.get(EXECUTOR_BIND_ADDRESS.key, executorHostname)
       val executorHostname = container.getNodeId.getHost
       val containerId = container.getId
       val executorId = executorIdCounter.toString
       val yarnResourceForRpId = rpIdToYarnResource.get(rpId)
       assert(container.getResource.getMemorySize >= yarnResourceForRpId.getMemorySize)
-      logInfo(s"Launching container $containerId on host $executorHostname " +
+      logInfo(s"Launching container $containerId on host $executorHostname ($executorBindAddress) " +
         s"for executor with ID $executorId for ResourceProfile Id $rpId")
 
       val rp = rpIdToResourceProfile(rpId)
@@ -763,6 +764,7 @@ private[yarn] class YarnAllocator(
                 sparkConf,
                 driverUrl,
                 executorId,
+                executorBindAddress,
                 executorHostname,
                 containerMem,
                 containerCores,