This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 311a4511ec56 [SPARK-58748][CORE][K8S] Preserve advertised driver host
for wildcard bind addresses
311a4511ec56 is described below
commit 311a4511ec5683f9af07515be78d0cfa1095a9af
Author: Chao Sun <[email protected]>
AuthorDate: Thu Aug 13 16:31:39 2026 -0700
[SPARK-58748][CORE][K8S] Preserve advertised driver host for wildcard bind
addresses
### What changes were proposed in this pull request?
Teach Spark to distinguish the address a Kubernetes driver **listens on**
from the address executors should **connect to** when direct driver-Pod-IP mode
is enabled.
Normally, using the driver's bind address as its advertised address works:
a Kubernetes-managed driver binds directly to its Pod IP, and executors can
connect to that IP without going through a driver Service. That assumption
breaks when a driver intentionally binds to every network interface. An address
such as `0.0.0.0` or `::` means "listen everywhere," but it does not identify a
reachable peer.
This change makes the choice depend on the address itself. A concrete bind
address continues to use the existing direct-Pod-IP path. A wildcard bind
address instead preserves the separately configured, routable
`spark.driver.host`. Both driver initialization and executor endpoint
construction apply the same rule, so the address advertised by the driver and
the address given to executors remain consistent. Wildcard recognition supports
IPv4 and IPv6 without performing DNS lookups, and [...]
### Why are the changes needed?
[SPARK-58748](https://issues.apache.org/jira/browse/SPARK-58748)
Spark deliberately separates two network settings:
```properties
# Where the driver opens its listening sockets.
spark.driver.bindAddress=0.0.0.0
# The reachable address advertised to executors.
spark.driver.host=10.0.0.42
```
This is a legitimate and common configuration for Kubernetes client-mode
applications and Spark Connect servers: the process listens on all local
interfaces, while executors are told to connect to the driver's actual Pod IP
or a routable Service hostname.
The problem appears when direct driver-Pod-IP mode is enabled:
```properties
spark.master=k8s://https://kubernetes.example:6443
spark.driver.bindAddress=0.0.0.0
spark.driver.host=10.0.0.42
spark.kubernetes.executor.useDriverPodIP=true
```
Spark currently assumes the driver's bind address is always its Pod IP.
Consequently, `SparkContext` replaces the configured advertised host with
`0.0.0.0`, and Kubernetes executor creation independently uses that same
wildcard to construct the scheduler endpoint:
```text
Before
Driver listens on: 0.0.0.0:7078
Advertised driver host: 0.0.0.0
Executor driver URL: spark://CoarseGrainedScheduler0.0.0.0:7078
Result: executors cannot connect or register
After
Driver listens on: 0.0.0.0:7078
Advertised driver host: 10.0.0.42
Executor driver URL: spark://CoarseGrainedScheduler10.0.0.42:7078
Result: executors connect normally
```
The driver starts successfully because binding to `0.0.0.0` is valid. The
failure only becomes visible when executors try to register, leaving the
application running but unable to execute tasks. Fixing only one side is
insufficient: the driver and executor each make their own address-selection
decision, and both must preserve the advertised host.
The same distinction applies to IPv6. For example, a driver may listen on
`::` while advertising a concrete address such as `[2001:db8::42]`; executors
must receive the concrete address, never the IPv6 wildcard. Compressed,
bracketed, and expanded IPv6 wildcard forms are handled consistently.
Spark 4.1 and 4.2 are affected when
`spark.kubernetes.executor.useDriverPodIP` is explicitly enabled. Spark 4.3 and
current master enable it by default, so previously valid wildcard-bind
configurations can fail without any application-level configuration change.
### Does this PR introduce _any_ user-facing change?
Yes. Kubernetes applications that bind their driver to all interfaces now
retain the reachable address configured in `spark.driver.host`, allowing their
executors to register and run. This restores the existing documented
distinction between the driver's listening address and its advertised address.
Applications that bind directly to a concrete Pod IP continue using
direct-IP routing, including existing IPv6 normalization. Applications with
direct driver-Pod-IP mode disabled, and applications outside Kubernetes, retain
their existing behavior. No configuration defaults or public APIs change.
### How was this patch tested?
```bash
build/sbt -Pkubernetes \
'core/testOnly org.apache.spark.SparkContextSuite -- -z "SPARK-58748"' \
'core/testOnly org.apache.spark.util.UtilsSuite -- -z "SPARK-58748"' \
'kubernetes/testOnly
org.apache.spark.deploy.k8s.features.BasicExecutorFeatureStepSuite' \
'core/scalastyle' \
'core/Test/scalastyle' \
'kubernetes/scalastyle' \
'kubernetes/Test/scalastyle'
```
All 41 tests passed: one targeted `SparkContextSuite` regression, one
targeted `UtilsSuite` regression, and all 39 `BasicExecutorFeatureStepSuite`
tests. Together, they cover default-enabled driver initialization, explicitly
enabled and disabled executor configuration, IPv4 and IPv6 wildcard addresses,
concrete IPv4 and IPv6 addresses, executor driver URLs, and existing IPv6
normalization. All four Scala style checks also passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #57977 from
sunchao/dev/chao/codex/spark-k8s-wildcard-driver-address-oss.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
---
.../main/scala/org/apache/spark/SparkContext.scala | 15 +++++--
.../main/scala/org/apache/spark/util/Utils.scala | 6 +++
.../scala/org/apache/spark/SparkContextSuite.scala | 46 +++++++++++++++++++++-
.../scala/org/apache/spark/util/UtilsSuite.scala | 9 +++++
.../k8s/features/BasicExecutorFeatureStep.scala | 3 +-
.../features/BasicExecutorFeatureStepSuite.scala | 11 ++++--
6 files changed, 82 insertions(+), 8 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 65e5a3c16a5f..1d27dea2b3f4 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -452,9 +452,18 @@ class SparkContext(config: SparkConf) extends Logging {
// instead of relying on the default value of the config constant.
if (SparkMasterRegex.isK8s(master) &&
_conf.getBoolean("spark.kubernetes.executor.useDriverPodIP", true)) {
- logInfo("Use DRIVER_BIND_ADDRESS instead of DRIVER_HOST_ADDRESS as
driver address " +
- "because spark.kubernetes.executor.useDriverPodIP is true in K8s
mode.")
- _conf.set(DRIVER_HOST_ADDRESS,
Utils.normalizeIpIfNeeded(_conf.get(DRIVER_BIND_ADDRESS)))
+ val bindAddress = _conf.get(DRIVER_BIND_ADDRESS)
+ if (Utils.isAnyLocalAddress(bindAddress)) {
+ val driverHost = _conf.get(DRIVER_HOST_ADDRESS)
+ logInfo(log"spark.kubernetes.executor.useDriverPodIP is true but bind
address " +
+ log"${MDC(LogKeys.BIND_ADDRESS, bindAddress)} is a wildcard; " +
+ log"preserving advertised driver host ${MDC(LogKeys.HOST,
driverHost)}")
+ _conf.set(DRIVER_HOST_ADDRESS, driverHost)
+ } else {
+ logInfo("Use DRIVER_BIND_ADDRESS instead of DRIVER_HOST_ADDRESS as
driver address " +
+ "because spark.kubernetes.executor.useDriverPodIP is true in K8s
mode.")
+ _conf.set(DRIVER_HOST_ADDRESS, Utils.normalizeIpIfNeeded(bindAddress))
+ }
} else {
_conf.set(DRIVER_HOST_ADDRESS, _conf.get(DRIVER_HOST_ADDRESS))
}
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala
b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 02c6e61067a3..8507c889c64b 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -960,6 +960,12 @@ private[spark] object Utils
}
}
+ /** Returns whether a literal IPv4 or IPv6 address binds to every local
interface. */
+ private[spark] def isAnyLocalAddress(host: String): Boolean = {
+ val address = host.stripPrefix("[").stripSuffix("]")
+ InetAddresses.isInetAddress(address) &&
InetAddresses.forString(address).isAnyLocalAddress
+ }
+
/**
* Checks if the host contains only valid hostname/ip without port
* NOTE: Incase of IPV6 ip it should be enclosed inside []
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index ef2d18f51306..e461d7624f62 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -45,7 +45,7 @@ import org.apache.spark.launcher.SparkLauncher
import org.apache.spark.resource.ResourceAllocation
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.resource.TestResourceIDs._
-import org.apache.spark.scheduler.{SparkListener,
SparkListenerExecutorMetricsUpdate, SparkListenerJobStart,
SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.scheduler.{LiveListenerBus, SparkListener,
SparkListenerExecutorMetricsUpdate, SparkListenerJobStart,
SparkListenerTaskEnd, SparkListenerTaskStart}
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.util.{ThreadUtils, Utils}
import org.apache.spark.util.ArrayImplicits._
@@ -713,6 +713,50 @@ class SparkContextSuite extends SparkFunSuite with
LocalSparkContext with Eventu
}
}
+ test("SPARK-58748: Kubernetes drivers do not advertise wildcard bind
addresses") {
+ Seq(
+ ("0.0.0.0", "10.129.36.37", "10.129.36.37"),
+ ("::", "10.129.36.37", "10.129.36.37"),
+ ("10.138.148.230", "driver-service", "10.138.148.230"),
+ ("2001:DB8:0:0::BEEF", "driver-service", "[2001:db8::beef]")).foreach {
+ case (bindAddress, advertisedAddress, expectedAddress) =>
+ var observedAddress: Option[String] = None
+ val logAppender = new LogAppender("wildcard driver bind address")
+ val conf = new SparkConf(false)
+ .setMaster("k8s://https://localhost:6443";)
+ .setAppName("driver-bind-address")
+ .set(DRIVER_BIND_ADDRESS, bindAddress)
+ .set(DRIVER_HOST_ADDRESS, advertisedAddress)
+
+ withLogAppender(logAppender) {
+ val error = intercept[SparkException] {
+ new SparkContext(conf) {
+ override private[spark] def createSparkEnv(
+ conf: SparkConf,
+ isLocal: Boolean,
+ listenerBus: LiveListenerBus): SparkEnv = {
+ observedAddress = Some(conf.get(DRIVER_HOST_ADDRESS))
+ throw new SparkException("stop after resolving the driver
address")
+ }
+ }
+ }
+ assert(error.getMessage === "stop after resolving the driver
address")
+ }
+
+ assert(observedAddress.contains(expectedAddress))
+ val wildcardMessages = logAppender.loggingEvents
+ .map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("is a wildcard; preserving advertised driver
host"))
+ if (Utils.isAnyLocalAddress(bindAddress)) {
+ assert(wildcardMessages.size === 1)
+ assert(wildcardMessages.head.contains(bindAddress))
+ assert(wildcardMessages.head.contains(advertisedAddress))
+ } else {
+ assert(wildcardMessages.isEmpty)
+ }
+ }
+ }
+
testCancellingTasks("that raise interrupted exception on cancel") {
Thread.sleep(9999999)
}
diff --git a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
index e1e49a226735..c467b7f0189c 100644
--- a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
@@ -1486,6 +1486,15 @@ class UtilsSuite extends SparkFunSuite with
ResetSystemProperties {
assert(Utils.buildLocationMetadata(paths, 18) == "(5 paths)[path0, path1,
...]")
}
+ test("SPARK-58748: wildcard IPv4 and IPv6 bind addresses are not advertised
driver addresses") {
+ Seq("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0").foreach { address =>
+ assert(Utils.isAnyLocalAddress(address), s"$address should be a wildcard
address")
+ }
+ Seq("10.129.36.37", "::1", "[::1]", "2001:db8::1", "localhost").foreach {
address =>
+ assert(!Utils.isAnyLocalAddress(address), s"$address should not be a
wildcard address")
+ }
+ }
+
test("checkHost supports both IPV4 and IPV6") {
// IPV4 ips
Utils.checkHost("0.0.0.0")
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
index 9eb9303fa0dd..af6273c131d4 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
@@ -53,7 +53,8 @@ private[spark] class BasicExecutorFeatureStep(
private val executorPodNamePrefix = kubernetesConf.resourceNamePrefix
- private val driverAddress = if
(kubernetesConf.get(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP)) {
+ private val driverAddress = if
(kubernetesConf.get(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP) &&
+ !Utils.isAnyLocalAddress(kubernetesConf.get(DRIVER_BIND_ADDRESS))) {
kubernetesConf.get(DRIVER_BIND_ADDRESS)
} else {
kubernetesConf.get(DRIVER_HOST_ADDRESS)
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
index a5a05b5fa8fe..70d5f6339892 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
@@ -352,11 +352,16 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite
with BeforeAndAfter {
ENV_EXECUTOR_ATTRIBUTE_EXECUTOR_ID -> KubernetesTestConf.EXECUTOR_ID))
}
- test("SPARK-53944: Support spark.kubernetes.executor.useDriverPodIP") {
+ test("SPARK-53944, SPARK-58748: Support
spark.kubernetes.executor.useDriverPodIP") {
Seq(
- (false, "bindAddress", "localhost"),
+ (false, "10.138.148.230", "localhost"),
+ (true, "10.138.148.230", "10.138.148.230"),
(true, "bindAddress", "bindAddress"),
- (true, "2001:DB8:0:0::BEEF", "[2001:db8::beef]")).foreach {
+ (true, "2001:DB8:0:0::BEEF", "[2001:db8::beef]"),
+ (true, "0.0.0.0", "localhost"),
+ (true, "::", "localhost"),
+ (true, "[::]", "localhost"),
+ (true, "0:0:0:0:0:0:0:0", "localhost")).foreach {
case (flag, bindAddress, address) =>
val conf = baseConf.clone()
.set(DRIVER_BIND_ADDRESS, bindAddress)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
