This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new cd847951973f [SPARK-58719][CORE][K8S] Fix hexadecimal IPv6 RPC address
normalization
cd847951973f is described below
commit cd847951973f6c3158754b1459d4480e99be7ab2
Author: Chao Sun <[email protected]>
AuthorDate: Wed Aug 12 09:38:55 2026 -0700
[SPARK-58719][CORE][K8S] Fix hexadecimal IPv6 RPC address normalization
### Why are the changes needed?
`RpcAddress` normalizes IPv6 literals so equivalent addresses compare
equally, but `Utils.normalizeIpIfNeeded` currently recognizes only decimal
digits and colons. Consequently, ordinary hexadecimal IPv6 addresses such as
`2001:db8::dead:beef` bypass normalization and are embedded in Spark RPC URLs
without the required brackets.
For example, enabling `spark.kubernetes.executor.useDriverPodIP` with the
driver pod address `2001:DB8:0:0::BEEF` currently produces an invalid executor
driver URL:
```
spark://CoarseGrainedScheduler2001:DB8:0:0::BEEF:7098
```
The correct canonical URL is:
```
spark://CoarseGrainedScheduler[2001:db8::beef]:7098
```
This also affects other RPC addresses that use IPv6 literals. The gap was
introduced when SPARK-42173 added IPv6 canonicalization without accepting
hexadecimal letters in its address matcher.
### What changes were proposed in this pull request?
- Accept uppercase and lowercase hexadecimal digits in both portions of the
existing IPv6 address matcher, preserving the existing Guava canonicalization
behavior.
- Extend `RpcAddressSuite` to cover raw and bracketed IPv6 literals,
uppercase and zero-padded representations, trailing hexadecimal segments,
equivalent-address equality, and Spark URL round trips.
- Extend the existing Kubernetes `spark.kubernetes.executor.useDriverPodIP`
regression test to verify that a raw IPv6 driver pod address produces a
correctly bracketed and canonical executor driver URL.
### Does this PR introduce _any_ user-facing change?
Yes. Valid hexadecimal IPv6 RPC addresses are now normalized and bracketed
correctly, allowing Kubernetes executors to connect directly to IPv6 driver pod
addresses. Existing IPv4, hostname, and numeric-only IPv6 behavior is unchanged.
### How was this patch tested?
```bash
build/sbt -Pkubernetes \
'core/testOnly org.apache.spark.rpc.RpcAddressSuite' \
'kubernetes/testOnly
org.apache.spark.deploy.k8s.features.BasicExecutorFeatureStepSuite'
```
- `RpcAddressSuite`: 11 tests passed.
- `BasicExecutorFeatureStepSuite`: 39 tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #57938 from sunchao/dev/chao/codex/ipv6-rpc-driver-pod-ip-oss.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
(cherry picked from commit 8064aa51e75cf787465618f9c901fc25f07c9f82)
Signed-off-by: Chao Sun <[email protected]>
---
.../main/scala/org/apache/spark/SparkContext.scala | 2 +-
.../main/scala/org/apache/spark/util/Utils.scala | 5 ++--
.../org/apache/spark/rpc/RpcAddressSuite.scala | 16 +++++++++++
.../features/BasicExecutorFeatureStepSuite.scala | 9 ++++--
.../k8s/KubernetesClusterManagerSuite.scala | 32 ++++++++++++++++++++++
5 files changed, 58 insertions(+), 6 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 96b326e107af..65e5a3c16a5f 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -454,7 +454,7 @@ class SparkContext(config: SparkConf) extends Logging {
_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, _conf.get(DRIVER_BIND_ADDRESS))
+ _conf.set(DRIVER_HOST_ADDRESS,
Utils.normalizeIpIfNeeded(_conf.get(DRIVER_BIND_ADDRESS)))
} 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 39f9ca8a472c..89001923f8c9 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -971,11 +971,12 @@ private[spark] object Utils
*/
private[spark] def normalizeIpIfNeeded(host: String): String = {
// Is this a v6 address. We ask users to add [] around v6 addresses as
strs but
- // there not always there. If it's just 0-9 and : and [] we treat it as a
v6 address.
+ // they're not always there. If it's just hexadecimal digits, :, and [] we
treat it as a
+ // v6 address.
// This means some invalid addresses are treated as v6 addresses, but
since they are
// not valid hostnames it doesn't matter.
// See https://www.rfc-editor.org/rfc/rfc1123#page-13 for context around
valid hostnames.
- val addressRe = """^\[{0,1}([0-9:]+?:[0-9]*)\]{0,1}$""".r
+ val addressRe = """^\[{0,1}([0-9a-fA-F:]+?:[0-9a-fA-F]*)\]{0,1}$""".r
host match {
case addressRe(unbracketed) =>
addBracketsIfNeeded(InetAddresses.toAddrString(InetAddresses.forString(unbracketed)))
diff --git a/core/src/test/scala/org/apache/spark/rpc/RpcAddressSuite.scala
b/core/src/test/scala/org/apache/spark/rpc/RpcAddressSuite.scala
index 9fb08c79420c..203810b7ca4a 100644
--- a/core/src/test/scala/org/apache/spark/rpc/RpcAddressSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rpc/RpcAddressSuite.scala
@@ -80,4 +80,20 @@ class RpcAddressSuite extends SparkFunSuite {
val address = RpcAddress("2600::", 1234)
assert(address.toSparkURL == "spark://[2600::]:1234")
}
+
+ test("SPARK-58719: Normalize hexadecimal IPv6 addresses") {
+ val expected = RpcAddress("2001:db8::dead:beef", 1234)
+ Seq(
+ "2001:db8::dead:beef",
+ "[2001:db8::dead:beef]",
+ "2001:0DB8:0000::DEAD:BEEF",
+ "[2001:0DB8:0000::DEAD:BEEF]").foreach { host =>
+ val address = RpcAddress(host, 1234)
+ assert(address.host == "[2001:db8::dead:beef]")
+ assert(address.hostPort == "[2001:db8::dead:beef]:1234")
+ assert(address.toSparkURL == "spark://[2001:db8::dead:beef]:1234")
+ assert(address == expected)
+ assert(RpcAddress.fromSparkURL(address.toSparkURL) == 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 ba5aa35985da..a5a05b5fa8fe 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
@@ -353,10 +353,13 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite
with BeforeAndAfter {
}
test("SPARK-53944: Support spark.kubernetes.executor.useDriverPodIP") {
- Seq((false, "localhost"), (true, "bindAddress")).foreach {
- case (flag, address) =>
+ Seq(
+ (false, "bindAddress", "localhost"),
+ (true, "bindAddress", "bindAddress"),
+ (true, "2001:DB8:0:0::BEEF", "[2001:db8::beef]")).foreach {
+ case (flag, bindAddress, address) =>
val conf = baseConf.clone()
- .set(DRIVER_BIND_ADDRESS, "bindAddress")
+ .set(DRIVER_BIND_ADDRESS, bindAddress)
.set(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP, flag)
val kconf = KubernetesTestConf.createExecutorConf(sparkConf = conf)
val step = new BasicExecutorFeatureStep(kconf, new
SecurityManager(conf), defaultProfile)
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManagerSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManagerSuite.scala
index 78e0942cfb82..55eb692d8ec7 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManagerSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManagerSuite.scala
@@ -16,6 +16,11 @@
*/
package org.apache.spark.scheduler.cluster.k8s
+import java.net.{InetSocketAddress, StandardProtocolFamily}
+import java.nio.channels.ServerSocketChannel
+
+import scala.util.Using
+
import io.fabric8.kubernetes.client.KubernetesClient
import org.mockito.{Mock, MockitoAnnotations}
import org.mockito.Mockito.when
@@ -28,6 +33,7 @@ import org.apache.spark.internal.config._
import org.apache.spark.scheduler.TaskSchedulerImpl
import
org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SPARK_APP_ID
import org.apache.spark.scheduler.local.LocalSchedulerBackend
+import org.apache.spark.util.RpcUtils
class KubernetesClusterManagerSuite extends SparkFunSuite with BeforeAndAfter {
@@ -88,6 +94,32 @@ class KubernetesClusterManagerSuite extends SparkFunSuite
with BeforeAndAfter {
assert(backend2.applicationId() === "user-app-id")
}
+ test("SPARK-58719: normalize IPv6 driver host when using the driver pod IP")
{
+ assume(
+ Using(ServerSocketChannel.open(StandardProtocolFamily.INET6)) { channel
=>
+ channel.bind(new InetSocketAddress("::1", 0))
+ }.isSuccess,
+ "IPv6 loopback is unavailable")
+
+ val rawAddress = "0:0:0:0:0:0:0:1"
+ val conf = new SparkConf(false)
+ .setAppName("ipv6-driver-host")
+ .setMaster("k8s://test")
+ .set(KUBERNETES_DRIVER_MASTER_URL, "local[2]")
+ .set(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP, true)
+ .set(DRIVER_BIND_ADDRESS, rawAddress)
+ .set("spark.ui.enabled", "false")
+
+ LocalSparkContext.withSpark(new SparkContext(conf)) { context =>
+ assert(context.conf.get(DRIVER_BIND_ADDRESS) === rawAddress)
+ assert(context.conf.get(DRIVER_HOST_ADDRESS) === "[::1]")
+ assert(context.env.blockManager.blockManagerId.host === "[::1]")
+ val driverRef = RpcUtils.makeDriverRef(
+ HeartbeatReceiver.ENDPOINT_NAME, context.conf, context.env.rpcEnv)
+ assert(driverRef.address.host === "[::1]")
+ }
+ }
+
test("deployment allocator with dynamic allocation requires deletion cost") {
val manager = new KubernetesClusterManager()
sparkConf.set(KUBERNETES_ALLOCATION_PODS_ALLOCATOR, "deployment")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
