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 5f084cba3041 [SPARK-58915][CORE][K8S][YARN] Add `supportsExecutorHold`
to the scheduler backends
5f084cba3041 is described below
commit 5f084cba3041d9104b89535153dc1f83468675fe
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Aug 20 13:29:42 2026 -0700
[SPARK-58915][CORE][K8S][YARN] Add `supportsExecutorHold` to the scheduler
backends
### What changes were proposed in this pull request?
This PR adds `private[spark]
CoarseGrainedSchedulerBackend.supportsExecutorHold`, which tells
whether a backend can hold its executors gracefully: publish an all-zero
executor requirement to
the cluster manager, and let the executors that are already running finish
their tasks and exit on
their own instead of being terminated.
It is `false` by default, so a backend has to opt in:
| Backend | Value | Reason |
|---|---|---|
| `CoarseGrainedSchedulerBackend` | `false` | `doRequestTotalExecutors`
does not acknowledge the published requirement |
| `StandaloneSchedulerBackend` | `true` | The Master honors a zero executor
limit without killing the running executors |
| `YarnSchedulerBackend` | `true` | The AM honors a zero executor target
without killing the running executors |
| `KubernetesClusterSchedulerBackend` | `true` (with the default `direct`
allocator only). | The other allocators depends on K8s management. |
### Why are the changes needed?
This is split out of https://github.com/apache/spark/pull/58054, which
holds and resumes an
application through graceful executor decommission. Whether a cluster
manager keeps its running
executors alive on a zero requirement is a per-resource-manager judgement,
orthogonal to the rest
of that PR and spanning Standalone, YARN, and Kubernetes, so it is reviewed
on its own.
### Does this PR introduce _any_ user-facing change?
No because this is a new `private[spark]` method.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #58174 from dongjoon-hyun/SPARK-58915.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 18c5d1e9dc2a688862d36c8f3b8d5ca815dc4800)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scheduler/cluster/CoarseGrainedSchedulerBackend.scala | 11 +++++++++++
.../scheduler/cluster/StandaloneSchedulerBackend.scala | 4 ++++
.../cluster/k8s/KubernetesClusterSchedulerBackend.scala | 8 ++++++++
.../k8s/KubernetesClusterSchedulerBackendSuite.scala | 14 ++++++++++++++
.../spark/scheduler/cluster/YarnSchedulerBackend.scala | 4 ++++
5 files changed, 41 insertions(+)
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
index 2f957aead71d..ad6ee1f81423 100644
---
a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
+++
b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -754,6 +754,17 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
executorDataMap.keySet.toSeq
}
+ /**
+ * Whether this backend can hold its executors gracefully: publish an
all-zero executor
+ * requirement to the cluster manager, and let the executors that are
already running finish
+ * their tasks and exit on their own instead of being terminated.
+ *
+ * False by default so that a backend has to opt in: the base
`doRequestTotalExecutors` does
+ * not acknowledge the published requirement at all, and a cluster manager
that terminates
+ * running executors once the requirement drops to zero cannot drain them
gracefully.
+ */
+ private[spark] def supportsExecutorHold: Boolean = false
+
def getExecutorsWithRegistrationTs(): Map[String, Long] = synchronized {
executorDataMap.toMap.transform((_, v) => v.registrationTs)
}
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
index 061b54914c83..751c4def7c92 100644
---
a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
+++
b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
@@ -248,6 +248,10 @@ private[spark] class StandaloneSchedulerBackend(
}
}
+ // The Master honors a zero executor limit without killing the running
executors, so the
+ // executors can be held gracefully.
+ private[spark] override def supportsExecutorHold: Boolean = true
+
/**
* Kill the given list of executors through the Master.
* @return whether the kill request is acknowledged.
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
index 255d76fece56..27a0b320cbf9 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
@@ -195,6 +195,14 @@ private[spark] class KubernetesClusterSchedulerBackend(
Future.successful(true)
}
+ // The Deployment/StatefulSet allocators (and unknown custom ones) scale
their controller
+ // down on a zero requirement, which deletes running executor pods after the
termination
+ // grace period instead of letting them finish their tasks, so holding is
supported only
+ // with the direct allocator.
+ private[spark] override def supportsExecutorHold: Boolean = {
+ conf.get(KUBERNETES_ALLOCATION_PODS_ALLOCATOR) == "direct"
+ }
+
override def sufficientResourcesRegistered(): Boolean = {
totalRegisteredExecutors.get() >= minRegisteredExecutors
}
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
index 5ac0d66f4183..64734cc6c612 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
@@ -496,4 +496,18 @@ class KubernetesClusterSchedulerBackendSuite extends
SparkFunSuite with BeforeAn
assert(id1 === id2, "applicationId() must return the same value on
repeated calls")
assert(id1.startsWith("spark-"), "generated app ID should have the spark-
prefix")
}
+
+ test("SPARK-58915: the executors can be held only with the direct pods
allocator") {
+ assert(schedulerBackendUnderTest.supportsExecutorHold)
+ Seq("statefulset", "deployment", "com.example.CustomAllocator").foreach {
allocator =>
+ sparkConf.set(KUBERNETES_ALLOCATION_PODS_ALLOCATOR, allocator)
+ try {
+ assert(!schedulerBackendUnderTest.supportsExecutorHold,
+ s"holding must not be supported with the $allocator allocator")
+ } finally {
+ sparkConf.remove(KUBERNETES_ALLOCATION_PODS_ALLOCATOR.key)
+ }
+ }
+ assert(schedulerBackendUnderTest.supportsExecutorHold)
+ }
}
diff --git
a/resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
b/resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
index b51c52e0e169..efe63895efeb 100644
---
a/resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
+++
b/resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
@@ -163,6 +163,10 @@ private[spark] abstract class YarnSchedulerBackend(
yarnSchedulerEndpointRef.ask[Boolean](prepareRequestExecutors(resourceProfileToTotalExecs))
}
+ // The AM honors a zero executor target without killing the running
executors, so the
+ // executors can be held gracefully.
+ private[spark] override def supportsExecutorHold: Boolean = true
+
/**
* Request that the ApplicationMaster kill the specified executors.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
