This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 fad330593475 [SPARK-58113][K8S] Support `publishNotReadyAddresses` in
driver service
fad330593475 is described below
commit fad330593475f53196b2dcdd4a4b9cb2f445553b
Author: Jiwon Park <[email protected]>
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 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 01b3c2534fb4afd5eef6ffafcaed1787558cf28e)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
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 390e054434ca..b88a313861ca 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: [email protected]
For additional commands, e-mail: [email protected]
