You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/07/15 14:31:17 UTC
(spark) branch master updated: [SPARK-58113][K8S] Support `publishNotReadyAddresses` in driver service
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 01b3c2534fb4 [SPARK-58113][K8S] Support `publishNotReadyAddresses` in driver service
01b3c2534fb4 is described below
commit 01b3c2534fb4afd5eef6ffafcaed1787558cf28e
Author: Jiwon Park <jp...@outlook.kr>
AuthorDate: Wed Jul 15 07:31:06 2026 -0700
[SPARK-58113][K8S] Support `publishNotReadyAddresses` in driver service
### What changes were proposed in this pull request?
Add a new configuration `spark.kubernetes.driver.service.publishNotReadyAddresses` (boolean, default `false`).
When enabled:
- `DriverServiceFeatureStep` sets `spec.publishNotReadyAddresses: true` on the headless driver service, so DNS records for the driver pod are published even while the pod is not Ready.
- The driver pod readiness wait before executor allocation (introduced in SPARK-32975) is skipped in `ExecutorPodsAllocator`, `StatefulSetPodsAllocator`, and `DeploymentPodsAllocator`, since the sole purpose of that wait — ensuring the headless service is resolvable by DNS before executors start — no longer depends on pod readiness.
### Why are the changes needed?
Kubernetes publishes DNS records for a headless service only for Ready pods. Users commonly attach a readiness probe to the driver pod through `spark.kubernetes.driver.podTemplateFile` — a typical setup for long-running Spark Connect servers, where pod readiness gates traffic routing. When such a probe gates on a port that binds after `SparkContext` initialization completes (e.g. the Spark Connect gRPC port), executors launched during initialization cannot resolve the driver service a [...]
The existing mitigation from SPARK-32975 (`waitUntilReady` in the allocators, tunable via `spark.kubernetes.allocation.driver.readinessTimeout` per SPARK-49079) cannot help here: the wait runs inside `SparkContext` initialization, while such a probe can only pass after initialization completes, so the wait always burns the full timeout and the race resumes.
The driver service is a single-pod discovery endpoint rather than a load-balancing service, and `publishNotReadyAddresses` is the standard Kubernetes pattern for headless discovery services. There is currently no first-class way to set it — the service spec is not customizable via pod templates, and the only workarounds are re-implementing the service in a custom feature step (via `spark.kubernetes.driver.pod.excludedFeatureSteps`) or cluster-level admission webhooks. This follows the [...]
The configuration is opt-in (default `false`) because publishing not-ready addresses is not desirable for deployments that rely on endpoint readiness to gate traffic to the driver service (e.g. through a service mesh); flipping the default could be discussed separately.
### Does this PR introduce _any_ user-facing change?
No change by default. A new opt-in configuration is added and documented in `running-on-kubernetes.md`.
### How was this patch tested?
New unit tests:
- `DriverServiceFeatureStepSuite`: the default leaves the field `false`; enabling the configuration sets it to `true`.
- `ExecutorPodsAllocatorSuite`, `StatefulSetAllocatorSuite`, `DeploymentAllocatorSuite`: the driver pod readiness wait still runs by default, and enabling the configuration skips it (`waitUntilReady` is not invoked).
Also verified the behavior end-to-end on a real Kubernetes cluster by applying `publishNotReadyAddresses: true` to the driver service through an admission webhook (equivalent to what this configuration does): executors resolved the driver service and registered while the driver pod was still NotReady behind a Spark Connect port readiness probe, and a static-allocation application that previously failed on every attempt with exit code 11 started successfully.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Co-authored with Claude Opus.
Closes #57242 from j1wonpark/SPARK-58113-publish-not-ready.
Authored-by: Jiwon Park <jp...@outlook.kr>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/running-on-kubernetes.md | 11 +++++++++
.../scala/org/apache/spark/deploy/k8s/Config.scala | 12 ++++++++++
.../k8s/features/DriverServiceFeatureStep.scala | 5 +++-
.../cluster/k8s/DeploymentPodsAllocator.scala | 19 +++++++++------
.../cluster/k8s/ExecutorPodsAllocator.scala | 24 ++++++++++++-------
.../cluster/k8s/StatefulSetPodsAllocator.scala | 24 ++++++++++++-------
.../features/DriverServiceFeatureStepSuite.scala | 28 ++++++++++++++++++++++
.../cluster/k8s/DeploymentAllocatorSuite.scala | 18 +++++++++++++-
.../cluster/k8s/ExecutorPodsAllocatorSuite.scala | 19 ++++++++++++++-
.../cluster/k8s/StatefulSetAllocatorSuite.scala | 17 ++++++++++++-
10 files changed, 148 insertions(+), 29 deletions(-)
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index feeed013b33e..23bc44c46751 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -1583,6 +1583,17 @@ See the [configuration page](configuration.html) for information on Spark config
</td>
<td>3.4.0</td>
</tr>
+<tr>
+ <td><code>spark.kubernetes.driver.service.publishNotReadyAddresses</code></td>
+ <td><code>false</code></td>
+ <td>
+ If true, the driver service publishes DNS records for the driver pod even while the pod
+ is not ready, so executors can resolve the driver service during startup when a readiness
+ probe is configured on the driver pod. When enabled, the driver pod readiness wait before
+ executor allocation is skipped as well.
+ </td>
+ <td>4.3.0</td>
+</tr>
<tr>
<td><code>spark.kubernetes.securityContext.allowPrivilegeEscalation</code></td>
<td><code>false</code></td>
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 4ec27b6fab75..00a02f3cba6f 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -88,6 +88,18 @@ private[spark] object Config extends Logging {
.checkValues(Set("IPv4", "IPv6", "IPv4,IPv6", "IPv6,IPv4"))
.createWithDefault("IPv4")
+ val KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES =
+ ConfigBuilder("spark.kubernetes.driver.service.publishNotReadyAddresses")
+ .doc("If true, the driver service publishes DNS records for the driver pod even " +
+ "while the pod is not ready, so executors can resolve the driver service " +
+ "during startup when a readiness probe is configured on the driver pod. " +
+ "When enabled, the driver pod readiness wait before executor allocation " +
+ "is skipped as well.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
val KUBERNETES_DRIVER_OWN_PVC =
ConfigBuilder("spark.kubernetes.driver.ownPersistentVolumeClaim")
.doc("If true, driver pod becomes the owner of on-demand persistent volume claims " +
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala
index f66c062fd4eb..18e2172e5181 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala
@@ -21,7 +21,7 @@ import scala.jdk.CollectionConverters._
import io.fabric8.kubernetes.api.model.{HasMetadata, ServiceBuilder}
import org.apache.spark.deploy.k8s.{KubernetesDriverConf, SparkPod}
-import org.apache.spark.deploy.k8s.Config.{KUBERNETES_DNS_LABEL_NAME_MAX_LENGTH, KUBERNETES_DRIVER_SERVICE_IP_FAMILIES, KUBERNETES_DRIVER_SERVICE_IP_FAMILY_POLICY}
+import org.apache.spark.deploy.k8s.Config.{KUBERNETES_DNS_LABEL_NAME_MAX_LENGTH, KUBERNETES_DRIVER_SERVICE_IP_FAMILIES, KUBERNETES_DRIVER_SERVICE_IP_FAMILY_POLICY, KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES}
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.internal.{config, Logging}
@@ -42,6 +42,8 @@ private[spark] class DriverServiceFeatureStep(
kubernetesConf.sparkConf.get(KUBERNETES_DRIVER_SERVICE_IP_FAMILY_POLICY)
private val ipFamilies =
kubernetesConf.sparkConf.get(KUBERNETES_DRIVER_SERVICE_IP_FAMILIES).split(",").toList.asJava
+ private val publishNotReadyAddresses =
+ kubernetesConf.sparkConf.get(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES)
private val driverPort = kubernetesConf.sparkConf.getInt(
config.DRIVER_PORT.key, DEFAULT_DRIVER_PORT)
@@ -70,6 +72,7 @@ private[spark] class DriverServiceFeatureStep(
.endMetadata()
.withNewSpec()
.withClusterIP("None")
+ .withPublishNotReadyAddresses(publishNotReadyAddresses)
.withIpFamilyPolicy(ipFamilyPolicy)
.withIpFamilies(ipFamilies)
.withSelector(kubernetesConf.labels.asJava)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentPodsAllocator.scala
index 213d12301a9c..0ad34fe9fe69 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentPodsAllocator.scala
@@ -54,6 +54,9 @@ class DeploymentPodsAllocator(
private val driverPodReadinessTimeout = conf.get(KUBERNETES_ALLOCATION_DRIVER_READINESS_TIMEOUT)
+ private val shouldWaitForDriverReadiness =
+ !conf.get(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES)
+
private val namespace = conf.get(KUBERNETES_NAMESPACE)
private val kubernetesDriverPodName = conf.get(KUBERNETES_DRIVER_POD_NAME)
@@ -77,13 +80,15 @@ class DeploymentPodsAllocator(
applicationId: String,
schedulerBackend: KubernetesClusterSchedulerBackend): Unit = {
appId = applicationId
- driverPod.foreach { pod =>
- Utils.tryLogNonFatalError {
- kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withName(pod.getMetadata.getName)
- .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ if (shouldWaitForDriverReadiness) {
+ driverPod.foreach { pod =>
+ Utils.tryLogNonFatalError {
+ kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .withName(pod.getMetadata.getName)
+ .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ }
}
}
}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
index c20346edc840..fb1206d6d102 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
@@ -88,6 +88,9 @@ class ExecutorPodsAllocator(
protected val driverPodReadinessTimeout = conf.get(KUBERNETES_ALLOCATION_DRIVER_READINESS_TIMEOUT)
+ private val shouldWaitForDriverReadiness =
+ !conf.get(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES)
+
protected val executorIdleTimeout = conf.get(DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT) * 1000
protected val namespace = conf.get(KUBERNETES_NAMESPACE)
@@ -132,15 +135,18 @@ class ExecutorPodsAllocator(
def start(applicationId: String, schedulerBackend: KubernetesClusterSchedulerBackend): Unit = {
appId = applicationId
- driverPod.foreach { pod =>
- // Wait until the driver pod is ready before starting executors, as the headless service won't
- // be resolvable by DNS until the driver pod is ready.
- Utils.tryLogNonFatalError {
- kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withName(pod.getMetadata.getName)
- .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ if (shouldWaitForDriverReadiness) {
+ driverPod.foreach { pod =>
+ // Wait until the driver pod is ready before starting executors, as the headless service
+ // won't be resolvable by DNS until the driver pod is ready. This is unnecessary when the
+ // driver service publishes not-ready addresses, since DNS no longer depends on readiness.
+ Utils.tryLogNonFatalError {
+ kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .withName(pod.getMetadata.getName)
+ .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ }
}
}
snapshotsStore.addSubscriber(podAllocationDelay) { executorPodsSnapshot =>
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala
index 0285ae396d6e..c3cb8cf97c34 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala
@@ -46,6 +46,9 @@ class StatefulSetPodsAllocator(
protected val driverPodReadinessTimeout = conf.get(KUBERNETES_ALLOCATION_DRIVER_READINESS_TIMEOUT)
+ private val shouldWaitForDriverReadiness =
+ !conf.get(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES)
+
protected val namespace = conf.get(KUBERNETES_NAMESPACE)
protected val kubernetesDriverPodName = conf
@@ -64,15 +67,18 @@ class StatefulSetPodsAllocator(
def start(applicationId: String, schedulerBackend: KubernetesClusterSchedulerBackend): Unit = {
appId = applicationId
- driverPod.foreach { pod =>
- // Wait until the driver pod is ready before starting executors, as the headless service won't
- // be resolvable by DNS until the driver pod is ready.
- Utils.tryLogNonFatalError {
- kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withName(pod.getMetadata.getName)
- .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ if (shouldWaitForDriverReadiness) {
+ driverPod.foreach { pod =>
+ // Wait until the driver pod is ready before starting executors, as the headless service
+ // won't be resolvable by DNS until the driver pod is ready. This is unnecessary when the
+ // driver service publishes not-ready addresses, since DNS no longer depends on readiness.
+ Utils.tryLogNonFatalError {
+ kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .withName(pod.getMetadata.getName)
+ .waitUntilReady(driverPodReadinessTimeout, TimeUnit.SECONDS)
+ }
}
}
}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala
index 57003571272e..ca42fbe68d34 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala
@@ -232,6 +232,34 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
}
}
+ test("SPARK-58113: publishNotReadyAddresses defaults to false") {
+ val kconf = KubernetesTestConf.createDriverConf(
+ sparkConf = new SparkConf(false),
+ labels = DRIVER_LABELS,
+ serviceLabels = DRIVER_SERVICE_LABELS,
+ serviceAnnotations = DRIVER_SERVICE_ANNOTATIONS)
+ val driverService = new DriverServiceFeatureStep(kconf)
+ .getAdditionalKubernetesResources()
+ .head
+ .asInstanceOf[Service]
+ assert(driverService.getSpec.getPublishNotReadyAddresses === false)
+ }
+
+ test("SPARK-58113: Support publishNotReadyAddresses") {
+ val sparkConf = new SparkConf(false)
+ .set(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES, true)
+ val kconf = KubernetesTestConf.createDriverConf(
+ sparkConf = sparkConf,
+ labels = DRIVER_LABELS,
+ serviceLabels = DRIVER_SERVICE_LABELS,
+ serviceAnnotations = DRIVER_SERVICE_ANNOTATIONS)
+ val driverService = new DriverServiceFeatureStep(kconf)
+ .getAdditionalKubernetesResources()
+ .head
+ .asInstanceOf[Service]
+ assert(driverService.getSpec.getPublishNotReadyAddresses === true)
+ }
+
private def verifyService(
driverPort: Int,
blockManagerPort: Int,
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentAllocatorSuite.scala
index 2166cef9d73c..0423e18ecd48 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentAllocatorSuite.scala
@@ -24,7 +24,7 @@ import io.fabric8.kubernetes.client.KubernetesClient
import io.fabric8.kubernetes.client.dsl.{AppsAPIGroupDSL, PodResource}
import org.mockito.{ArgumentCaptor, Mock, MockitoAnnotations}
import org.mockito.ArgumentMatchers.{any, eq => meq}
-import org.mockito.Mockito.{never, times, verify, when}
+import org.mockito.Mockito.{clearInvocations, never, times, verify, when}
import org.scalatest.BeforeAndAfter
import org.apache.spark.{SecurityManager, SparkConf, SparkException, SparkFunSuite}
@@ -115,6 +115,22 @@ class DeploymentAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
ResourceProfile.clearDefaultProfile()
}
+ test("SPARK-58113: wait for driver readiness by default") {
+ // The allocator in `before` was started with the default conf.
+ verify(driverPodResource, times(1)).waitUntilReady(any(), any())
+ }
+
+ test("SPARK-58113: skip driver readiness wait when publishNotReadyAddresses is enabled") {
+ clearInvocations(driverPodResource)
+ val confWithPublishNotReady = conf.clone()
+ .set(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES, true)
+ val podsAllocator = new DeploymentPodsAllocator(
+ confWithPublishNotReady, secMgr, executorBuilder, kubernetesClient, snapshotsStore,
+ snapshotsStore.clock)
+ podsAllocator.start(TEST_SPARK_APP_ID, schedulerBackend)
+ verify(driverPodResource, never()).waitUntilReady(any(), any())
+ }
+
test("creates deployments per resource profile and seeds deletion cost annotation") {
val rpBuilder = new ResourceProfileBuilder()
val secondProfile = rpBuilder.build()
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
index 843e704e5c01..3522c49682ab 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
@@ -28,7 +28,7 @@ import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientException
import io.fabric8.kubernetes.client.dsl.PodResource
import org.mockito.{Mock, MockitoAnnotations}
import org.mockito.ArgumentMatchers.{any, anyString, eq => meq}
-import org.mockito.Mockito.{never, times, verify, when}
+import org.mockito.Mockito.{clearInvocations, never, times, verify, when}
import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
import org.scalatest.BeforeAndAfter
@@ -895,6 +895,23 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
" namespace default"))
}
+ test("SPARK-58113: wait for driver readiness by default") {
+ // The allocator in `before` was started with the default conf.
+ verify(driverPodOperations, times(1)).waitUntilReady(any(), any())
+ }
+
+ test("SPARK-58113: skip driver readiness wait when publishNotReadyAddresses is enabled") {
+ clearInvocations(driverPodOperations)
+ val confWithPublishNotReady = conf.clone()
+ .set(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES, true)
+ val podsAllocator = new ExecutorPodsAllocator(
+ confWithPublishNotReady, secMgr, executorBuilder, kubernetesClient, snapshotsStore,
+ waitForExecutorPodsClock)
+ podsAllocator.setExecutorPodsLifecycleManager(lifecycleManager)
+ podsAllocator.start(TEST_SPARK_APP_ID, schedulerBackend)
+ verify(driverPodOperations, never()).waitUntilReady(any(), any())
+ }
+
test("SPARK-39688: getReusablePVCs should handle accounts with no PVC permission") {
val getReusablePVCs =
PrivateMethod[mutable.Buffer[PersistentVolumeClaim]](Symbol("getReusablePVCs"))
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetAllocatorSuite.scala
index 474d5b060ebf..9d96cf546c39 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetAllocatorSuite.scala
@@ -22,7 +22,7 @@ import io.fabric8.kubernetes.client.KubernetesClient
import io.fabric8.kubernetes.client.dsl._
import org.mockito.{ArgumentCaptor, Mock, MockitoAnnotations}
import org.mockito.ArgumentMatchers.{any, eq => meq}
-import org.mockito.Mockito.{never, times, verify, when}
+import org.mockito.Mockito.{clearInvocations, never, times, verify, when}
import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
import org.scalatest.BeforeAndAfter
@@ -128,6 +128,21 @@ class StatefulSetAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
podsAllocatorUnderTest.start(TEST_SPARK_APP_ID, schedulerBackend)
}
+ test("SPARK-58113: wait for driver readiness by default") {
+ // The allocator in `before` was started with the default conf.
+ verify(driverPodOperations, times(1)).waitUntilReady(any(), any())
+ }
+
+ test("SPARK-58113: skip driver readiness wait when publishNotReadyAddresses is enabled") {
+ clearInvocations(driverPodOperations)
+ val confWithPublishNotReady = conf.clone()
+ .set(KUBERNETES_DRIVER_SERVICE_PUBLISH_NOT_READY_ADDRESSES, true)
+ val podsAllocator = new StatefulSetPodsAllocator(
+ confWithPublishNotReady, secMgr, executorBuilder, kubernetesClient, snapshotsStore, null)
+ podsAllocator.start(TEST_SPARK_APP_ID, schedulerBackend)
+ verify(driverPodOperations, never()).waitUntilReady(any(), any())
+ }
+
test("Validate initial statefulSet creation & cleanup with two resource profiles") {
val rprof = new ResourceProfileBuilder()
val taskReq = new TaskResourceRequests().resource("gpu", 1)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org