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 6b6bc512e72a [SPARK-58828][CORE] Support holding and resuming an
application via graceful executor decommission
6b6bc512e72a is described below
commit 6b6bc512e72abcf64facc60405c1a34f7d295dde
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Aug 20 16:24:28 2026 -0700
[SPARK-58828][CORE] Support holding and resuming an application via
graceful executor decommission
### What changes were proposed in this pull request?
This PR adds the ability to hold and resume a whole application from the
Driver Web UI.
- New `DeveloperApi` methods `SparkContext.holdExecutors()` /
`resumeExecutors()`. Holding gracefully decommissions all existing executors,
so each finishes its running tasks and then exits (unless
`spark.executor.decommission.forceKillTimeout` is set, which kills a still-busy
executor after that timeout), and makes the scheduler backend publish an
all-zero executor requirement while keeping the requested totals recorded
(`CoarseGrainedSchedulerBackend.publishTotals()`; `ExecutorAlloc [...]
- The Jobs page shows an application status line with a `(hold)` /
`(resume)` control and the outcome of the last request, following the existing
kill button pattern, gated by a new config `spark.ui.holdEnabled` (default
`true`).
- Scheduler interactions: a barrier job submitted while held does not
consume its retry budget (`DAGScheduler`). Pipelined-shuffle jobs and workloads
with long-running tasks (streaming receivers, continuous processing) are
outside the hold's scope: a hold is rejected while a pipelined job is running
(`TaskSchedulerImpl.hasPipelinedTaskSets`), a pipelined job submitted while
held fails its gang admission immediately and should be resubmitted after the
resume (with the internal `spark.s [...]
This requires `spark.decommission.enabled`, shuffle data kept outside the
executors (either an external shuffle service via
`spark.shuffle.service.enabled` or a `ShuffleDataIO` with reliable storage),
and a scheduler backend that opts into holding
(`CoarseGrainedSchedulerBackend.supportsExecutorHold`: Standalone, YARN, and
Kubernetes with `spark.kubernetes.allocation.pods.allocator=direct`). It is
rejected otherwise.
### Why are the changes needed?
To allow a running application to temporarily give back its resources
without being killed. Executors are drained gracefully with no loss of
in-progress task work or shuffle output, and the application can be resumed
later without recomputing the shuffle output already written. Cached blocks are
not preserved and are recomputed after resuming.
```mermaid
stateDiagram-v2
state "Running (hold)" as running
state "Held (draining 1 executor) (resume)" as draining
state "Held (resume)" as held
[*] --> running
running --> draining : click hold (decommission all executors)
draining --> held : running tasks finish, executors exit
held --> running : click resume (restore executor targets)
```
<img width="770" height="748" alt="Screenshot 2026-08-17 at 15 25 03"
src="https://github.com/user-attachments/assets/67f0887a-9d08-4400-bf83-c6ce3256a0c0";
/>
<img width="770" height="748" alt="Screenshot 2026-08-17 at 15 25 30"
src="https://github.com/user-attachments/assets/0617da2a-287b-480c-af57-352524f76e47";
/>
<img width="770" height="748" alt="Screenshot 2026-08-17 at 15 25 50"
src="https://github.com/user-attachments/assets/ddc9f9b3-f7dd-454b-9e75-faaee310c8a4";
/>
<img width="770" height="748" alt="Screenshot 2026-08-17 at 15 26 58"
src="https://github.com/user-attachments/assets/811b90b1-9250-4605-8733-255097e636b6";
/>
### Does this PR introduce _any_ user-facing change?
No, this is a new feature.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #58054 from dongjoon-hyun/SPARK-58828.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 0d3a4317bad5e2df9756b74f16d3ea179a22e605)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/ExecutorAllocationManager.scala | 145 ++++++++++++-
.../main/scala/org/apache/spark/SparkContext.scala | 237 ++++++++++++++++++++-
.../org/apache/spark/internal/config/UI.scala | 13 ++
.../org/apache/spark/scheduler/DAGScheduler.scala | 18 +-
.../cluster/CoarseGrainedSchedulerBackend.scala | 142 +++++++++++-
.../main/scala/org/apache/spark/ui/SparkUI.scala | 11 +-
.../org/apache/spark/ui/jobs/AllJobsPage.scala | 25 +++
.../scala/org/apache/spark/ui/jobs/JobsTab.scala | 90 ++++++++
.../spark/ExecutorAllocationManagerSuite.scala | 141 ++++++++++++
.../scala/org/apache/spark/SparkContextSuite.scala | 153 ++++++++++++-
.../CoarseGrainedSchedulerBackendSuite.scala | 90 ++++++++
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 26 +++
.../org/apache/spark/ui/UISeleniumSuite.scala | 23 ++
docs/configuration.md | 15 ++
docs/web-ui.md | 18 ++
15 files changed, 1133 insertions(+), 14 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index 559b3fcf3712..fe6a3c0d1bf6 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -69,7 +69,9 @@ import org.apache.spark.util.{Clock, SystemClock,
ThreadUtils, Utils}
* blocks will be removed if it has been idle for more than L seconds.
*
* There is no retry logic in either case because we make the assumption that
the cluster manager
- * will eventually fulfill all requests it receives asynchronously.
+ * will eventually fulfill all requests it receives asynchronously. The one
exception is the
+ * deferred target push used while the executors are held (see
`targetSyncPending`), which
+ * retries until the cluster manager acknowledges the current targets.
*
* The relevant Spark properties are below. Each of these properties applies
separately to
* every ResourceProfile. So if you set a minimum number of executors, that is
a minimum
@@ -175,6 +177,28 @@ private[spark] class ExecutorAllocationManager(
// (2) an executor idle timeout has elapsed.
@volatile private var initializing: Boolean = true
+ // Whether allocation is suspended because the executors are held. While
this is true,
+ // `schedule()` is a no-op so that pending tasks do not bring up new
executors.
+ // See `SparkContext.holdExecutors()`.
+ private var suspended: Boolean = false
+
+ // Whether the current executor targets still have to be pushed to the
cluster manager. Set
+ // when a push from `suspend()`/`resume()` fails or is rejected (e.g. before
the YARN AM has
+ // registered), and by `reset()`, which may run inside a cluster manager RPC
handler where a
+ // synchronous request would self-deadlock (e.g. YARN's
RegisterClusterManager). The push is
+ // performed from the allocation thread in `schedule()` and retried until
acknowledged, with
+ // an exponential backoff: the conditions under which it retries (an AM
restart, an
+ // unreachable cluster manager) resolve on a timescale far above the
allocation tick.
+ private var targetSyncPending: Boolean = false
+
+ // Ticks to wait before the next deferred push attempt, and the delay to arm
after another
+ // failure. Zero keeps the first attempt after arming `targetSyncPending`
immediate.
+ private var ticksUntilTargetSync: Int = 0
+ private var targetSyncBackoffTicks: Int = 0
+
+ // Upper bound of the deferred push backoff, in ticks (10 seconds with the
100ms interval).
+ private val maxTargetSyncBackoffTicks = 100
+
// Number of locality aware tasks for each ResourceProfile, used for
executor placement.
private var numLocalityAwareTasksPerResourceProfileId = new
mutable.HashMap[Int, Int]
numLocalityAwareTasksPerResourceProfileId(defaultProfileId) = 0
@@ -282,12 +306,107 @@ private[spark] class ExecutorAllocationManager(
def reset(): Unit = synchronized {
addTime = 0L
numExecutorsTargetPerResourceProfileId.keys.foreach { rpId =>
- numExecutorsTargetPerResourceProfileId(rpId) = initialNumExecutors
+ numExecutorsTargetPerResourceProfileId(rpId) = if (suspended) 0 else
initialNumExecutors
}
numExecutorsToAddPerResourceProfileId.keys.foreach { rpId =>
numExecutorsToAddPerResourceProfileId(rpId) = 1
}
executorMonitor.reset()
+ if (suspended) {
+ // A restarted cluster manager AM may have allocated executors on its
own, so the zero
+ // targets have to be pushed again. Leave that to `schedule()`: this
method may run
+ // inside the cluster manager's RPC handler, where a synchronous request
would
+ // self-deadlock.
+ targetSyncPending = true
+ ticksUntilTargetSync = 0
+ targetSyncBackoffTicks = 0
+ }
+ }
+
+ /**
+ * Push the current executor targets to the cluster manager and report
whether the request
+ * was acknowledged, arming the `schedule()` retry otherwise.
+ */
+ private def syncTargetsWithClient(): Boolean = {
+ val acknowledged = try {
+ client.requestTotalExecutors(
+ numExecutorsTargetPerResourceProfileId.toMap,
+ numLocalityAwareTasksPerResourceProfileId.toMap,
+ rpIdToHostToLocalTaskCount)
+ } catch {
+ case NonFatal(e) =>
+ // Use INFO level to be consistent with `doUpdateRequest`: errors here
are more
+ // commonly caused by YARN AM restarts, which is a recoverable issue.
+ logInfo("Error reaching cluster manager.", e)
+ false
+ }
+ if (acknowledged) {
+ targetSyncPending = false
+ ticksUntilTargetSync = 0
+ targetSyncBackoffTicks = 0
+ } else {
+ targetSyncPending = true
+ ticksUntilTargetSync = targetSyncBackoffTicks
+ targetSyncBackoffTicks =
+ math.min(maxTargetSyncBackoffTicks, math.max(1, targetSyncBackoffTicks
* 2))
+ }
+ acknowledged
+ }
+
+ /**
+ * Suspend allocation and lower the executor targets of all resource
profiles to zero, so that
+ * pending tasks do not bring up new executors while the executors are held.
Target updates
+ * are a no-op in `schedule()` until [[resume()]] is called.
+ *
+ * @return whether the zero targets were acknowledged by the cluster
manager. When false, the
+ * push is retried from `schedule()` until acknowledged.
+ */
+ def suspend(): Boolean = synchronized {
+ if (!suspended) {
+ suspended = true
+ numExecutorsTargetPerResourceProfileId.keys.foreach { rpId =>
+ numExecutorsTargetPerResourceProfileId(rpId) = 0
+ }
+ numExecutorsToAddPerResourceProfileId.keys.foreach { rpId =>
+ numExecutorsToAddPerResourceProfileId(rpId) = 1
+ }
+ syncTargetsWithClient()
+ } else {
+ true
+ }
+ }
+
+ /**
+ * Resume allocation suspended by [[suspend()]]. The executor targets are
recomputed from the
+ * current load on the next `schedule()` run.
+ *
+ * @return whether the restored targets were acknowledged by the cluster
manager. When false,
+ * the push is retried from `schedule()` until acknowledged.
+ */
+ def resume(): Boolean = synchronized {
+ if (suspended) {
+ suspended = false
+ // Restore at least the lower bound of the target: the initial warm-up
requested by
+ // `start()` when no stage has been submitted yet
(`updateAndSyncNumExecutorsTarget`
+ // skips it while initializing), and `minNumExecutors` otherwise, so
that an idle
+ // application does not stay below the minimum until the next backlog.
+ val floor = if (initializing) initialNumExecutors else minNumExecutors
+ numExecutorsTargetPerResourceProfileId.keys.foreach { rpId =>
+ numExecutorsTargetPerResourceProfileId(rpId) =
+ math.max(numExecutorsTargetPerResourceProfileId(rpId), floor)
+ }
+ val acknowledged = syncTargetsWithClient()
+ if (!initializing && numExecutorsTargetPerResourceProfileId.keys
+ .exists(maxNumExecutorsNeededPerResourceProfile(_) > 0)) {
+ // Trigger an immediate ramp-up for the load that built up while
suspended. Do not
+ // touch `addTime` otherwise: nothing resets it to NOT_SET while idle,
and a stale
+ // value would let a much later backlog skip the scheduler backlog
timeout.
+ addTime = clock.nanoTime()
+ }
+ acknowledged
+ } else {
+ true
+ }
}
/**
@@ -339,6 +458,21 @@ private[spark] class ExecutorAllocationManager(
* This is factored out into its own method for testing.
*/
private def schedule(): Unit = synchronized {
+ if (targetSyncPending) {
+ if (ticksUntilTargetSync <= 0) {
+ // Deferred target push, retried with a backoff until acknowledged: an
earlier push
+ // was rejected or failed (e.g. before the YARN AM registered), or
`reset()` ran
+ // inside a cluster manager RPC handler where a synchronous request
would
+ // self-deadlock. No `testing` short-circuit here: tests assert on
this call through
+ // the mocked client.
+ syncTargetsWithClient()
+ } else {
+ ticksUntilTargetSync -= 1
+ }
+ }
+ if (suspended) {
+ return
+ }
val executorIdsToBeRemoved = executorMonitor.timedOutExecutors()
if (executorIdsToBeRemoved.nonEmpty) {
initializing = false
@@ -717,8 +851,11 @@ private[spark] class ExecutorAllocationManager(
updateExecutorPlacementHints()
if (!numExecutorsTargetPerResourceProfileId.contains(profId)) {
- numExecutorsTargetPerResourceProfileId.put(profId,
initialNumExecutors)
- if (initialNumExecutors > 0) {
+ // While suspended, a new resource profile must start at a zero
target so that the
+ // hold is not bypassed; the target is recomputed on resume.
+ numExecutorsTargetPerResourceProfileId.put(
+ profId, if (suspended) 0 else initialNumExecutors)
+ if (!suspended && initialNumExecutors > 0) {
logDebug(s"requesting executors, rpId: $profId, initial number is
$initialNumExecutors")
// we need to trigger a schedule since we add an initial number
here.
client.requestTotalExecutors(
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 72830a565f80..438c036a359a 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -61,7 +61,7 @@ import org.apache.spark.resource._
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.rpc.RpcEndpointRef
import org.apache.spark.scheduler._
-import org.apache.spark.scheduler.cluster.StandaloneSchedulerBackend
+import org.apache.spark.scheduler.cluster.{CoarseGrainedSchedulerBackend,
SchedulerBackendUtils, StandaloneSchedulerBackend}
import org.apache.spark.scheduler.local.LocalSchedulerBackend
import org.apache.spark.shuffle.ShuffleDataIOUtils
import org.apache.spark.shuffle.api.ShuffleDriverComponents
@@ -245,6 +245,13 @@ class SparkContext(config: SparkConf) extends Logging {
private var _plugins: Option[PluginContainer] = None
private var _resourceProfileManager: ResourceProfileManager = _
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
Declared here so
+ // that the initializers run before the constructor exposes the context
through the UI:
+ // a hold served while `postStartHook()` is still waiting must not be erased.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
/*
-------------------------------------------------------------------------------------
*
| Accessors and public fields. These provide access to the internal state
of the |
| context.
|
@@ -2075,6 +2082,234 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors and can hold them,
decommission support,
+ * and shuffle data kept outside the executors: either an external shuffle
service or a
+ * `ShuffleDataIO` with reliable storage.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ (schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend => cg.supportsExecutorHold
+ case _ => false
+ }) &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
+ conf.get(DECOMMISSION_ENABLED)
+ }
+
+ /** Whether the executors are currently held by `holdExecutors()`. */
+ private[spark] def executorsHeld: Boolean = _executorsHeld
+
+ /**
+ * :: DeveloperApi ::
+ * Hold the whole application by declining to allocate new executors and
gracefully
+ * decommissioning all existing ones. Each executor finishes its running
tasks and then
+ * exits -- unless `spark.executor.decommission.forceKillTimeout` is set, in
which case an
+ * executor still running tasks is killed after that timeout. The shuffle
data already
+ * written remains available outside the executors, so the application can
later pick up
+ * where it left off via `resumeExecutors()`. Cached blocks are not
preserved and are
+ * recomputed after resuming. While held the application has no executors,
so with
+ * `spark.default.parallelism` unset the default parallelism falls back to
2, and an RDD
+ * created during the hold keeps that partition count after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`),
shuffle data kept
+ * outside the executors -- either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage -- and a
+ * scheduler backend that can hold executors: Standalone, YARN, and
Kubernetes with the
+ * `direct` pods allocator. Fallback storage
+ * (`spark.storage.decommission.fallbackStorage.path`) deliberately does not
qualify:
+ * shuffle blocks not yet migrated when an executor exits are dropped.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * Pipelined-shuffle jobs are outside the hold's scope. A hold is rejected,
on a
+ * best-effort check, while a pipelined job is running: its transient
shuffle data lives
+ * only on the executors and would not survive the drain, and a group that
slips past the
+ * check is aborted rather than drained, since a pipelined task set
tolerates no task
+ * failure. A pipelined job submitted while held fails its gang admission
immediately
+ * instead of waiting; resubmit it after the resume (with
+ * `spark.scheduler.pipelinedGroup.slotCheck.enabled=false` there is no
admission check,
+ * so it waits for the resume instead). Workloads with long-running tasks (a
+ * streaming receiver, continuous processing) are outside the scope too:
their tasks never
+ * finish, so the drain cannot complete.
+ *
+ * @throws IllegalArgumentException when the decommission or shuffle-storage
precondition is
+ * not met; an unsupported scheduler backend instead returns false
with a warning.
+ * @return whether the lowered executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background; the
+ * executors are drained in either case.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend if cg.supportsExecutorHold =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ val pipelinedRunning = taskScheduler match {
+ case ts: TaskSchedulerImpl => ts.hasPipelinedTaskSets
+ case _ => false
+ }
+ if (pipelinedRunning) {
+ // A pipelined group reads and writes transient shuffle data that
lives only on
+ // its executors: a partially launched group would deadlock the
drain, and a
+ // force-killed member aborts the whole group.
+ logWarning(log"Cannot hold the executors while a pipelined job is
running.")
+ false
+ } else synchronized {
+ if (_executorsHeld) {
+ // A repeated hold re-asserts the zero requirement: the earlier
publish may not
+ // have been acknowledged, and with dynamic allocation off nothing
retries it.
+ if (executorAllocationManager.isDefined) true else
zeroExecutorRequirementAndDrain(cg)
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume when none was explicitly
requested
+ // (explicitly requested totals, made before or during the hold,
are
+ // republished from the backend directly). Only killExecutors'
bookkeeping
+ // zero is kill-seeded (read atomically against a concurrent
reset): restore
+ // the count of executors not already being removed, so that
resume neither
+ // parks the application at zero nor undoes the downscale.
Otherwise
+ // Standalone has no explicit requirement by default (and ignores
+ // spark.executor.instances, even a leftover value), so restore
an unbounded
+ // one; elsewhere follow the conf, or fall back to the cluster
manager's
+ // default when no executor has registered yet.
+ heldNumExecutors = if (cg.hasKillSeededTotalsOnly) {
+ cg.activeExecutorCount
+ } else {
+ schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(cg.getExecutorIds().size,
+ SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS))
+ }
+ }
+ }
+ // Mark the hold before talking to the cluster manager, so that a
partial failure
+ // below leaves the executors held, and thus resumable, instead of
half-held.
+ _executorsHeld = true
+ cg.setExecutorsHeld(true)
+ executorAllocationManager match {
+ case Some(manager) =>
+ val acknowledged = manager.suspend()
+ drainHeldExecutors(cg)
+ acknowledged
+ case None => zeroExecutorRequirementAndDrain(cg)
+ }
+ }
+ }
+ case _ =>
+ logWarning("Holding executors is not supported by current scheduler.")
+ false
+ }
+ }
+
+ // Gracefully decommission all the current executors of a held application
and let the
+ // executor monitor know, so that it does not try to remove the draining
executors again and
+ // reports them in the decommissioning metrics.
+ private def drainHeldExecutors(b: ExecutorAllocationClient): Unit = {
+ val executors = b.getExecutorIds()
+ if (executors.nonEmpty) {
+ val decommissioned = b.decommissionExecutors(
+ executors.map(id => (id, ExecutorDecommissionInfo("Executors are
held"))).toArray,
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
+ }
+ }
+
+ // Restore the hold invariant without dynamic allocation: publish the zero
requirement
+ // (the requested totals are kept and republished on resume) and drain the
current
+ // executors. The publish must not abort the drain, which has to run even
when the cluster
+ // manager is temporarily unreachable; the registration guard remains the
backstop when the
+ // publish fails.
+ private[spark] def zeroExecutorRequirementAndDrain(
+ cg: CoarseGrainedSchedulerBackend): Boolean = {
+ val acknowledged = try {
+ cg.republishRequestedTotals()
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to lower the executor requirement while holding
the " +
+ log"executors.", e)
+ false
+ }
+ drainHeldExecutors(cg)
+ acknowledged
+ }
+
+ /**
+ * :: DeveloperApi ::
+ * Resume an application held by `holdExecutors()` by restoring its executor
requirements.
+ *
+ * @return whether the restored executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background and the
+ * hold is lifted; otherwise the executors stay held so the call can
be retried.
+ */
+ @DeveloperApi
+ def resumeExecutors(): Boolean = {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend =>
+ synchronized {
+ if (!_executorsHeld) {
+ true
+ } else {
+ // Lift the backend guard before restoring the requirement, so
that an executor
+ // granted by the restored requirement cannot race with its own
registration and
+ // be drained.
+ cg.setExecutorsHeld(false)
+ val acknowledged = executorAllocationManager match {
+ case Some(manager) =>
+ // resume() retries a rejected push in the background, so the
hold can be
+ // lifted regardless of the acknowledgment.
+ manager.resume()
+ case None =>
+ try {
+ // Totals requested before or during the hold are
republished as-is; the
+ // check and the publish are atomic, so a concurrent cluster
manager reset
+ // cannot turn this into publishing an empty map. When none
are recorded
+ // (never requested, or cleared by such a reset), restore
the requirement
+ // captured at hold.
+ cg.republishExplicitTotals().getOrElse {
+ // Publish without recording: the pre-hold state had no
explicitly
+ // requested totals, and recording the restore (in
particular
+ // Standalone's unbounded sentinel) would flip
killExecutors' empty-map
+ // seeding and mark the totals explicit for good.
+ cg.publishTotalsWithoutRecording(
+
immutable.Map(resourceProfileManager.defaultResourceProfile ->
+ heldNumExecutors))
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to restore the executor requirement
while resuming " +
+ log"the executors.", e)
+ false
+ }
+ }
+ if (executorAllocationManager.isDefined || acknowledged) {
+ _executorsHeld = false
+ } else {
+ // The requirement could not be restored: stay held and re-arm
the guard. The
+ // recorded totals are still the non-zero pre-hold ones, and an
executor may
+ // have registered while the guard was down, so restore the hold
invariant:
+ // push the zero requirement again and drain any current
executors. The call
+ // can be retried.
+ cg.setExecutorsHeld(true)
+ zeroExecutorRequirementAndDrain(cg)
+ logWarning(log"The cluster manager did not acknowledge the
restored executor " +
+ log"requirement; the executors remain held and
resumeExecutors() can be " +
+ log"retried.")
+ }
+ acknowledged
+ }
+ }
+ case _ =>
+ logWarning("Resuming executors is not supported by current scheduler.")
+ false
+ }
+ }
+
/** The version of Spark on which this application is running. */
def version: String = SPARK_VERSION
diff --git a/core/src/main/scala/org/apache/spark/internal/config/UI.scala
b/core/src/main/scala/org/apache/spark/internal/config/UI.scala
index 65f26e9ca279..e70743fa9df0 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/UI.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/UI.scala
@@ -92,6 +92,19 @@ private[spark] object UI {
.booleanConf
.createWithDefault(true)
+ val UI_HOLD_ENABLED = ConfigBuilder("spark.ui.holdEnabled")
+ .doc("Allows the whole application to be held and resumed from the web UI.
Holding " +
+ "gracefully decommissions all executors and stops requesting new ones.
Cached blocks " +
+ "are not preserved and are recomputed after resuming. This takes effect
only when " +
+ "spark.decommission.enabled is true, the shuffle data is kept outside
the executors " +
+ "(through either spark.shuffle.service.enabled or a ShuffleDataIO with
reliable " +
+ "storage), and the cluster manager can hold executors: Standalone, YARN,
and " +
+ "Kubernetes with spark.kubernetes.allocation.pods.allocator=direct.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val UI_THREAD_DUMPS_ENABLED = ConfigBuilder("spark.ui.threadDumpsEnabled")
.doc("Whether to show a link for executor thread dumps in Stages and
Executor pages.")
.version("1.2.0")
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index cdc65c035a60..fa2ab84c64e7 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -435,6 +435,13 @@ private[spark] class DAGScheduler(
matched.toSeq
}
+ /**
+ * Whether the executors are held (see `SparkContext.holdExecutors()`).
While held, the
+ * barrier slot check reads zero capacity, so its retry budget must not be
consumed: the job
+ * should wait for the resume. Extracted as a seam so tests can control the
hold state.
+ */
+ protected def executorsHeld: Boolean = sc.executorsHeld
+
/**
* Time in seconds to wait between a max concurrent tasks check failure and
the next check.
*/
@@ -2170,8 +2177,15 @@ private[spark] class DAGScheduler(
} catch {
case e: BarrierJobSlotsNumberCheckFailed =>
// If jobId doesn't exist in the map, Scala coverts its value null to
0: Int automatically.
- val numCheckFailures =
barrierJobIdToNumTasksCheckFailures.compute(jobId,
- (_: Int, value: Int) => value + 1)
+ // Do not consume the retry budget while the executors are held: the
slot check sees
+ // zero slots for the whole hold, and the job should wait for the
resume like any
+ // other job instead of failing when the retries run out.
+ val numCheckFailures = if (executorsHeld) {
+ barrierJobIdToNumTasksCheckFailures.getOrDefault(jobId, 0)
+ } else {
+ barrierJobIdToNumTasksCheckFailures.compute(jobId,
+ (_: Int, value: Int) => value + 1)
+ }
logWarning(log"Barrier stage in job ${MDC(JOB_ID, jobId)} " +
log"requires ${MDC(NUM_SLOTS, e.requiredConcurrentTasks)} slots, " +
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 ad6ee1f81423..1a2d72b9d741 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
@@ -107,6 +107,17 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
// Executors which are being decommissioned. Maps from executorId to
ExecutorDecommissionInfo.
protected val executorsPendingDecommission = new HashMap[String,
ExecutorDecommissionInfo]
+ // Whether the executors are held (see `SparkContext.holdExecutors()`).
While true, newly
+ // registered executors are decommissioned immediately, so that executors
the cluster manager
+ // granted before the hold cannot outlive it.
+ @volatile private var executorsHeld = false
+
+ // Whether an executor total was ever explicitly requested (through
requestExecutors or
+ // requestTotalExecutors), as opposed to the bookkeeping seed that
`adjustExecutors` writes
+ // when killing an executor that was started by default.
+ @GuardedBy("CoarseGrainedSchedulerBackend.this")
+ private var explicitExecutorRequest = false
+
// Unknown Executors which are being decommissioned. This could be caused by
unregistered executor
// This executor should be decommissioned after registration.
// Maps from executorId to (ExecutorDecommissionInfo,
adjustTargetNumExecutors,
@@ -321,6 +332,22 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
decommissionExecutors(Array((executorId, v._1)), v._2, v._3)
unknownExecutorsPendingDecommission.invalidate(executorId)
})
+ if (executorsHeld) {
+ // The executors are held; drain this late-registered executor
immediately. Note
+ // that ExecutorMonitor cannot be told here: it builds its state
asynchronously
+ // from the SparkListenerExecutorAdded event posted above, so this
executor is not
+ // tracked there yet and its decommissioning is under-reported in
the metrics.
+ // The monitor's own executor-removal handling covers the eventual
exit.
+ decommissionExecutors(
+ Array((executorId, ExecutorDecommissionInfo("Executors are
held"))),
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ // The cluster manager granted this executor against a stale
requirement (e.g. a
+ // restarted AM using its own initial target, or a lost response
to an earlier
+ // request), so re-assert the zero requirement, or it would keep
granting
+ // replacements that churn through this drain.
+ reassertHeldRequirement()
+ }
context.reply(true)
}
@@ -696,7 +723,19 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
* */
protected[scheduler] def reset(): Unit = {
val executors: Set[String] = synchronized {
- requestedTotalExecutorsPerResourceProfile.clear()
+ if (executorsHeld) {
+ // Keep the requested totals so that resume can restore them, and
re-assert the zero
+ // requirement so the restarted cluster manager AM does not allocate
executors
+ // against its own initial target while the executors are held. Do not
await the
+ // response: this may run inside the cluster manager's RPC handler.
+ publishTotals()
+ } else {
+ requestedTotalExecutorsPerResourceProfile.clear()
+ // Clear the explicit-request record with the totals it describes, so
that a later
+ // resume does not republish an empty map, which YARN would treat as
cancelling
+ // every target.
+ explicitExecutorRequest = false
+ }
executorDataMap.keys.toSet
}
@@ -765,6 +804,99 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
*/
private[spark] def supportsExecutorHold: Boolean = false
+ /** See `SparkContext.holdExecutors()`. */
+ private[spark] def setExecutorsHeld(held: Boolean): Unit = {
+ executorsHeld = held
+ }
+
+ /** Whether an executor total was ever explicitly requested. Visible for
testing only. */
+ private[spark] def hasExplicitExecutorRequests: Boolean = synchronized {
+ explicitExecutorRequest
+ }
+
+ /** The number of executors that are neither being removed nor
decommissioned. */
+ private[spark] def activeExecutorCount: Int = synchronized {
+ executorDataMap.keys.count(isExecutorActive)
+ }
+
+ /**
+ * Whether the tracked totals are only killExecutors' bookkeeping seed:
present, but never
+ * explicitly requested. Read atomically, so that a concurrent `reset()`
cannot split the
+ * decision.
+ */
+ private[spark] def hasKillSeededTotalsOnly: Boolean = synchronized {
+ !explicitExecutorRequest &&
requestedTotalExecutorsPerResourceProfile.nonEmpty
+ }
+
+ /**
+ * Republish the explicitly requested totals and await the acknowledgment,
or None when
+ * there are none: never explicitly requested, or cleared by a concurrent
`reset()`. The
+ * check and the publish are atomic, and an empty map is never published,
since some
+ * cluster managers treat it as cancelling every target.
+ */
+ private[spark] def republishExplicitTotals(): Option[Boolean] = {
+ val response = synchronized {
+ if (explicitExecutorRequest &&
requestedTotalExecutorsPerResourceProfile.nonEmpty) {
+ Some(publishTotals())
+ } else {
+ None
+ }
+ }
+ response.map(defaultAskTimeout.awaitResult(_))
+ }
+
+ /**
+ * Publish the given totals once, without recording them, and await the
acknowledgment.
+ * Used to restore a requirement the application never explicitly requested
(in particular
+ * Standalone's unbounded default on resume): recording it would flip
`adjustExecutors`'
+ * empty-map seeding and mark the totals explicit, changing how
`killExecutors` and a later
+ * hold behave for the rest of the application's life.
+ */
+ private[spark] def publishTotalsWithoutRecording(
+ totals: Map[ResourceProfile, Int]): Boolean = {
+ val response = synchronized {
+ doRequestTotalExecutors(totals)
+ }
+ defaultAskTimeout.awaitResult(response)
+ }
+
+ /**
+ * Re-assert the zero executor requirement of a held application without
awaiting the
+ * response and without touching the requested totals. The held flag is
re-checked under
+ * the lock: a concurrent `resumeExecutors()` lifts the flag before
restoring the
+ * requirement, so a stale re-assertion cannot overwrite a restored one.
+ */
+ private[spark] def reassertHeldRequirement(): Unit = synchronized {
+ if (executorsHeld) {
+ publishTotals()
+ }
+ }
+
+ /**
+ * Publish the current executor totals (all-zero while the executors are
held) and await
+ * the acknowledgment.
+ */
+ private[spark] def republishRequestedTotals(): Boolean = {
+ defaultAskTimeout.awaitResult(publishTotals())
+ }
+
+ /**
+ * Publish the executor totals to the cluster manager: the requested totals
normally, or
+ * all-zero totals while the executors are held, so that requirements
recorded during a
+ * hold are retained but nothing is allocated until resume.
+ *
+ * @return a future whose evaluation indicates whether the request is
acknowledged.
+ */
+ private def publishTotals(): Future[Boolean] = synchronized {
+ val totals = if (executorsHeld) {
+ requestedTotalExecutorsPerResourceProfile.map { case (rp, _) => (rp, 0)
}.toMap +
+ (scheduler.sc.resourceProfileManager.defaultResourceProfile -> 0)
+ } else {
+ requestedTotalExecutorsPerResourceProfile.toMap
+ }
+ doRequestTotalExecutors(totals)
+ }
+
def getExecutorsWithRegistrationTs(): Map[String, Long] = synchronized {
executorDataMap.toMap.transform((_, v) => v.registrationTs)
}
@@ -841,6 +973,7 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
log"executor(s) from the cluster manager")
val response = synchronized {
+ explicitExecutorRequest = true
val defaultProf =
scheduler.sc.resourceProfileManager.defaultResourceProfile
val numExisting =
requestedTotalExecutorsPerResourceProfile.getOrElse(defaultProf, 0)
// Saturate instead of overflowing when the current requirement is
already huge (e.g.
@@ -850,7 +983,7 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
requestedTotalExecutorsPerResourceProfile(defaultProf) = newTotal
// Account for executors pending to be added or removed
updateExecRequestTime(defaultProf.id, newTotal - numExisting)
- doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
+ publishTotals()
}
defaultAskTimeout.awaitResult(response)
@@ -888,6 +1021,7 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
(scheduler.sc.resourceProfileManager.resourceProfileFromId(rpid), num)
}
val response = synchronized {
+ explicitExecutorRequest = true
val oldResourceProfileToNumExecutors =
requestedTotalExecutorsPerResourceProfile.map {
case (rp, num) =>
(rp.id, num)
@@ -897,7 +1031,7 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
this.numLocalityAwareTasksPerResourceProfileId =
numLocalityAwareTasksPerResourceProfileId
this.rpHostToLocalTaskCount = hostToLocalTaskCount
updateExecRequestTimes(oldResourceProfileToNumExecutors,
resourceProfileIdToNumExecutors)
- doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
+ publishTotals()
}
defaultAskTimeout.awaitResult(response)
}
@@ -970,7 +1104,7 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
}
}
}
- doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
+ publishTotals()
} else {
Future.successful(true)
}
diff --git a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
index 862e150acd44..97c5d1687344 100644
--- a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
@@ -55,6 +55,8 @@ private[spark] class SparkUI private (
val killEnabled = sc.map(_.conf.get(UI_KILL_ENABLED)).getOrElse(false)
+ val holdEnabled = sc.map(_.conf.get(UI_HOLD_ENABLED)).getOrElse(false)
+
var appId: String = _
private var streamingJobProgressListener: Option[SparkListener] = None
@@ -94,9 +96,11 @@ private[spark] class SparkUI private (
}
}
+ private var jobsTab: JobsTab = _
+
/** Initialize all components of the server. */
def initialize(): Unit = {
- val jobsTab = new JobsTab(this, store)
+ jobsTab = new JobsTab(this, store)
attachTab(jobsTab)
val stagesTab = new StagesTab(this, store)
attachTab(stagesTab)
@@ -123,6 +127,10 @@ private[spark] class SparkUI private (
attachHandler(createRedirectHandler(
"/stages/stage/kill", "/stages/", stagesTab.handleKillRequest,
httpMethods = Set("GET", "POST")))
+ attachHandler(createRedirectHandler(
+ "/jobs/hold", "/jobs/", jobsTab.handleHoldRequest, httpMethods =
Set("GET", "POST")))
+ attachHandler(createRedirectHandler(
+ "/jobs/resume", "/jobs/", jobsTab.handleResumeRequest, httpMethods =
Set("GET", "POST")))
}
initialize()
@@ -164,6 +172,7 @@ private[spark] class SparkUI private (
/** Stop the server behind this web interface. Only valid after bind(). */
override def stop(): Unit = {
super.stop()
+ Option(jobsTab).foreach(_.stop())
logInfo(log"Stopped Spark web UI at ${MDC(WEB_URL, webUrl)}")
}
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
index 70d0a9faf4de..a0f9e52f2fb8 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
@@ -354,6 +354,31 @@ private[ui] class AllJobsPage(parent: JobsTab, store:
AppStatusStore) extends We
<strong>Scheduling Mode: </strong>
{schedulingMode}
</li>
+ {
+ if (parent.holdEnabled &&
parent.sc.exists(_.executorHoldSupported)) {
+ val basePathUri = UIUtils.prependBaseUri(request,
parent.basePath)
+ val (status, action, confirm) = if (parent.sc.get.executorsHeld)
{
+ val numDraining = parent.sc.get.getExecutorIds().size
+ val status = if (numDraining > 0) {
+ s"Held (draining $numDraining executor${if (numDraining > 1)
"s" else ""})"
+ } else {
+ "Held"
+ }
+ (status, "resume", "Are you sure you want to resume this
application?")
+ } else {
+ ("Running", "hold", "Are you sure you want to hold this
application? All " +
+ "executors will be decommissioned after finishing their
running tasks.")
+ }
+ <li>
+ <strong>Application:</strong>
+ {status}
+ <a href={s"$basePathUri/jobs/$action/"}
+ data-confirm-message={confirm}
+ class="confirm-link">{s"($action)"}</a>
+ {parent.lastHoldRequestStatus.getOrElse("")}
+ </li>
+ }
+ }
{
if (shouldShowActiveJobs) {
<li>
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
index a3c2d05414a8..98ae0de22fb3 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
@@ -17,6 +17,8 @@
package org.apache.spark.ui.jobs
+import java.util.concurrent.{ExecutorService, RejectedExecutionException}
+
import jakarta.servlet.http.HttpServletRequest
import org.apache.spark.JobExecutionStatus
@@ -24,6 +26,7 @@ import org.apache.spark.internal.config.SCHEDULER_MODE
import org.apache.spark.scheduler.SchedulingMode
import org.apache.spark.status.AppStatusStore
import org.apache.spark.ui._
+import org.apache.spark.util.{ThreadUtils, Utils}
/** Web UI showing progress status of all jobs in the given SparkContext. */
private[ui] class JobsTab(parent: SparkUI, store: AppStatusStore)
@@ -32,6 +35,7 @@ private[ui] class JobsTab(parent: SparkUI, store:
AppStatusStore)
val sc = parent.sc
val conf = parent.conf
val killEnabled = parent.killEnabled
+ val holdEnabled = parent.holdEnabled
// Show pool information for only live UI.
def isFairScheduler: Boolean = {
@@ -62,4 +66,90 @@ private[ui] class JobsTab(parent: SparkUI, store:
AppStatusStore)
}
}
}
+
+ // Serves the hold/resume requests off the Jetty serving thread: they talk
to the cluster
+ // manager and may block up to the RPC ask timeout. A single thread also
serializes
+ // concurrent requests. Created on first use and shut down by `stop()`, so
that the thread
+ // does not outlive the SparkContext.
+ private var holdRequestExecutor: Option[ExecutorService] = None
+ private var stopped = false
+
+ // None once stopped, so that a request served during teardown neither hits
a rejected
+ // execution on the shut-down pool nor recreates it and leaks the thread.
+ private def holdRequestExecutorPool: Option[ExecutorService] = synchronized {
+ if (stopped) {
+ None
+ } else {
+ Some(holdRequestExecutor.getOrElse {
+ val pool =
ThreadUtils.newDaemonSingleThreadExecutor("spark-ui-hold-resume")
+ holdRequestExecutor = Some(pool)
+ pool
+ })
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ stopped = true
+ holdRequestExecutor.foreach(_.shutdownNow())
+ }
+
+ // Outcome of the last hold/resume request served by this tab, as (isHold,
message): Some
+ // while a request is running or after it did not take effect, None when
idle or after a
+ // success. Rendered with the operation it belongs to, so a stale message
never reads as
+ // the opposite operation's, and a failed hold stays visible even though the
page already
+ // shows the (resume) control (the hold is marked before the cluster manager
is asked).
+ @volatile private var holdRequestStatus: Option[(Boolean, String)] = None
+
+ private[jobs] def lastHoldRequestStatus: Option[String] =
+ holdRequestStatus.map { case (isHold, message) =>
+ s"(${if (isHold) "hold" else "resume"}: $message)"
+ }
+
+ def handleHoldRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.foreach { pool =>
+ holdRequestStatus = Some((true, "requested"))
+ try {
+ pool.execute { () =>
+ var acknowledged = false
+ Utils.tryLogNonFatalError { acknowledged = ctx.holdExecutors() }
+ holdRequestStatus = if (acknowledged) {
+ None
+ } else {
+ Some((true, "the last request did not take effect, see the
driver logs"))
+ }
+ }
+ } catch {
+ // stop() may have shut the pool down after the accessor returned
it
+ case _: RejectedExecutionException =>
+ }
+ }
+ }
+ }
+ }
+
+ def handleResumeRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.foreach { pool =>
+ holdRequestStatus = Some((false, "requested"))
+ try {
+ pool.execute { () =>
+ var acknowledged = false
+ Utils.tryLogNonFatalError { acknowledged = ctx.resumeExecutors()
}
+ holdRequestStatus = if (acknowledged) {
+ None
+ } else {
+ Some((false, "the last request did not take effect, see the
driver logs"))
+ }
+ }
+ } catch {
+ // stop() may have shut the pool down after the accessor returned
it
+ case _: RejectedExecutionException =>
+ }
+ }
+ }
+ }
+ }
}
diff --git
a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
index 7f622dc91772..32289e74ccac 100644
--- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
@@ -1396,6 +1396,147 @@ class ExecutorAllocationManagerSuite extends
SparkFunSuite {
assert(numExecutorsTargetForDefaultProfileId(manager) === 20) // limit
reached
}
+ test("SPARK-58828: suspend and resume executor allocation") {
+ val clock = new ManualClock(2020L)
+ val manager = createManager(createConf(0, 20, 0), clock = clock)
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000)))
+
+ // Ramp up the target normally first
+ onSchedulerBacklogged(manager)
+ clock.advance(schedulerBacklogTimeout * 1000)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 1)
+ clock.advance(sustainedSchedulerBacklogTimeout * 1000)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 1 + 2)
+
+ // Suspending lowers the target to zero and stops the ramp-up
+ manager.suspend()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ clock.advance(sustainedSchedulerBacklogTimeout * 1000)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ clock.advance(sustainedSchedulerBacklogTimeout * 1000)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+
+ // Resuming ramps the target up again for the still-pending tasks
+ manager.resume()
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 1)
+ clock.advance(sustainedSchedulerBacklogTimeout * 1000)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 1 + 2)
+ }
+
+ test("SPARK-58828: suspend keeps a zero target across reset()") {
+ val manager = createManager(createConf(1, 10, 3))
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000)))
+ manager.suspend()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ manager.reset()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ manager.resume()
+ manager.reset()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 3)
+ }
+
+ test("SPARK-58828: resume before the first stage restores the initial
target") {
+ val manager = createManager(createConf(1, 10, 3))
+ manager.suspend()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ manager.resume()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 3)
+ }
+
+ test("SPARK-58828: a new resource profile submitted while suspended gets a
zero target") {
+ val manager = createManager(createConf(1, 10, 1))
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000, rp =
defaultProfile)))
+ manager.suspend()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ val rp1 = new ResourceProfileBuilder()
+ val execReqs = new ExecutorResourceRequests().cores(4).resource("gpu", 4)
+ val taskReqs = new TaskResourceRequests().cpus(1).resource("gpu", 1)
+ rp1.require(execReqs).require(taskReqs)
+ val rprof1 = rp1.build()
+ rpManager.addResourceProfile(rprof1)
+ post(SparkListenerStageSubmitted(createStageInfo(1, 1000, rp = rprof1)))
+ assert(numExecutorsTarget(manager, rprof1.id) === 0)
+ // After resume, the new profile is restored to the minimum and ramps up
like any other
+ manager.resume()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 1)
+ assert(numExecutorsTarget(manager, rprof1.id) === 1)
+ schedule(manager)
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 2)
+ assert(numExecutorsTarget(manager, rprof1.id) === 2)
+ }
+
+ test("SPARK-58828: resume restores at least the minimum target when idle") {
+ val manager = createManager(createConf(2, 10, 2))
+ post(SparkListenerStageSubmitted(createStageInfo(0, 8)))
+ post(SparkListenerStageCompleted(createStageInfo(0, 8)))
+ manager.suspend()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 0)
+ manager.resume()
+ assert(numExecutorsTargetForDefaultProfileId(manager) === 2)
+ }
+
+ test("SPARK-58828: a rejected suspend push is retried from the schedule
loop") {
+ val manager = createManager(createConf(1, 10, 3))
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000)))
+ when(client.requestTotalExecutors(any(), any(), any())).thenReturn(false)
+ clearInvocations(client)
+ assert(!manager.suspend())
+ verify(client).requestTotalExecutors(mockitoEq(Map(defaultProfile.id ->
0)), any(), any())
+ // Rejected: re-pushed on every tick until acknowledged, then no more
+ when(client.requestTotalExecutors(any(), any(), any())).thenReturn(true)
+ clearInvocations(client)
+ schedule(manager)
+ verify(client).requestTotalExecutors(mockitoEq(Map(defaultProfile.id ->
0)), any(), any())
+ clearInvocations(client)
+ schedule(manager)
+ verify(client, never()).requestTotalExecutors(any(), any(), any())
+ }
+
+ test("SPARK-58828: deferred target pushes back off exponentially") {
+ val manager = createManager(createConf(1, 10, 3))
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000)))
+ when(client.requestTotalExecutors(any(), any(), any())).thenReturn(false)
+ clearInvocations(client)
+ assert(!manager.suspend()) // first push fails; the next retry is immediate
+ verify(client).requestTotalExecutors(any(), any(), any())
+ clearInvocations(client)
+ schedule(manager) // immediate retry fails; the next retry waits a tick
+ verify(client).requestTotalExecutors(any(), any(), any())
+ clearInvocations(client)
+ schedule(manager) // backing off
+ verify(client, never()).requestTotalExecutors(any(), any(), any())
+ schedule(manager) // retried after the backoff
+ verify(client).requestTotalExecutors(any(), any(), any())
+ }
+
+ test("SPARK-58828: reset while suspended defers the zero-target push to the
schedule loop") {
+ // Keep DYN_ALLOCATION_TESTING on: turning it off would start the real
polling thread
+ // (`start()` ignores TEST_DYNAMIC_ALLOCATION_SCHEDULE_ENABLED when not
testing) and its
+ // background ticks would race with the assertions below.
+ val manager = createManager(createConf(1, 10, 3))
+ when(client.requestTotalExecutors(any(), any(), any())).thenReturn(true)
+ post(SparkListenerStageSubmitted(createStageInfo(0, 1000)))
+ manager.suspend()
+ clearInvocations(client)
+ // reset() itself must not talk to the cluster manager: it may run inside
a cluster
+ // manager RPC handler (e.g. YARN's RegisterClusterManager) where a
synchronous request
+ // would self-deadlock.
+ manager.reset()
+ verify(client, never()).requestTotalExecutors(any(), any(), any())
+ // The zero target is pushed from the schedule loop instead, exactly once
+ schedule(manager)
+ verify(client).requestTotalExecutors(mockitoEq(Map(defaultProfile.id ->
0)), any(), any())
+ clearInvocations(client)
+ schedule(manager)
+ verify(client, never()).requestTotalExecutors(any(), any(), any())
+ }
+
test("mock polling loop remove behavior") {
val clock = new ManualClock(2020L)
val manager = createManager(createConf(1, 20, 1), clock = clock)
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index e461d7624f62..f7d668377c1c 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark
import java.io.File
import java.net.{MalformedURLException, URI}
import java.nio.file.Files
-import java.util.concurrent.{CountDownLatch, Semaphore, TimeUnit}
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, Semaphore,
TimeUnit}
import scala.concurrent.duration._
import scala.io.Source
@@ -33,6 +33,8 @@ import org.apache.hadoop.mapred.TextInputFormat
import org.apache.hadoop.mapreduce.lib.input.{TextInputFormat =>
NewTextInputFormat}
import org.apache.logging.log4j.{Level, LogManager}
import org.json4s.{DefaultFormats, Extraction}
+import org.mockito.ArgumentMatchers.{any, eq => meq}
+import org.mockito.Mockito.{mock, verify, when}
import org.scalatest.concurrent.Eventually
import org.scalatest.matchers.must.Matchers._
@@ -42,10 +44,14 @@ import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Tests._
import org.apache.spark.internal.config.UI._
import org.apache.spark.launcher.SparkLauncher
+import org.apache.spark.network.TransportContext
+import org.apache.spark.network.netty.SparkTransportConf
+import org.apache.spark.network.shuffle.ExternalBlockHandler
import org.apache.spark.resource.ResourceAllocation
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.resource.TestResourceIDs._
-import org.apache.spark.scheduler.{LiveListenerBus, SparkListener,
SparkListenerExecutorMetricsUpdate, SparkListenerJobStart,
SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.scheduler.{LiveListenerBus, SparkListener,
SparkListenerExecutorMetricsUpdate, SparkListenerJobStart,
SparkListenerStageSubmitted, SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.util.{ThreadUtils, Utils}
import org.apache.spark.util.ArrayImplicits._
@@ -1578,6 +1584,149 @@ class SparkContextSuite extends SparkFunSuite with
LocalSparkContext with Eventu
assert(err.getMessage.contains("Int.MaxValue"))
assert(err.getMessage.contains("overflowed"))
}
+
+ test("SPARK-58828: holdExecutors and resumeExecutors are unsupported by the
local scheduler") {
+ sc = new SparkContext(new
SparkConf().setAppName("test").setMaster("local"))
+ assert(!sc.executorHoldSupported)
+ assert(!sc.holdExecutors())
+ assert(!sc.resumeExecutors())
+ }
+
+ test("SPARK-58828: holdExecutors requires external shuffle service and
decommission support") {
+ sc = new SparkContext(
+ new SparkConf().setAppName("test").setMaster("local-cluster[1,1,1024]"))
+ assert(!sc.executorHoldSupported)
+ val err = intercept[IllegalArgumentException] {
+ sc.holdExecutors()
+ }
+ assert(err.getMessage.contains(SHUFFLE_SERVICE_ENABLED.key))
+ assert(err.getMessage.contains(DECOMMISSION_ENABLED.key))
+ }
+
+ private def withExternalShuffleServer(conf: SparkConf)(body: => Unit): Unit
= {
+ // The executors register with the external shuffle service on startup, so
run one
+ val transportConf = SparkTransportConf.fromSparkConf(conf, "shuffle",
numUsableCores = 2)
+ val rpcHandler = new ExternalBlockHandler(transportConf, null)
+ val transportContext = new TransportContext(transportConf, rpcHandler)
+ val server = transportContext.createServer()
+ try {
+ conf.set(SHUFFLE_SERVICE_PORT, server.getPort)
+ body
+ } finally {
+ Utils.tryLogNonFatalError(server.close())
+ Utils.tryLogNonFatalError(rpcHandler.close())
+ Utils.tryLogNonFatalError(transportContext.close())
+ }
+ }
+
+ private def verifyHoldAndResumeExecutors(conf: SparkConf): Unit = {
+ withExternalShuffleServer(conf) {
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 1, 60000)
+ assert(sc.executorHoldSupported)
+ assert(!sc.executorsHeld)
+
+ // Shuffle output written before the hold must survive the drain
+ val shuffled = sc.parallelize(1 to 100, 4).map(i => (i % 8,
i)).reduceByKey(_ + _)
+ assert(shuffled.count() === 8)
+ val shuffleId =
+ shuffled.dependencies.head.asInstanceOf[ShuffleDependency[_, _,
_]].shuffleId
+ val tracker =
sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
+ assert(tracker.getNumAvailableOutputs(shuffleId) === 4)
+
+ assert(sc.holdExecutors())
+ assert(sc.executorsHeld)
+ // The executor finishes decommissioning and exits, and no new one
replaces it
+ eventually(timeout(60.seconds)) {
+ assert(sc.getExecutorIds().isEmpty)
+ }
+ // And stays drained: a transiently empty poll would also pass the check
above under a
+ // register-and-drain churn, so verify the zero requirement actually
settled
+ Thread.sleep(2000)
+ assert(sc.getExecutorIds().isEmpty)
+
+ assert(sc.resumeExecutors())
+ assert(!sc.executorsHeld)
+ // The restored requirement brings an executor back
+ eventually(timeout(60.seconds)) {
+ assert(sc.getExecutorIds().nonEmpty)
+ }
+
+ // The map output survived the drain: only the reduce stage re-runs when
the shuffle is
+ // read again. Asserted after the job, since the executor removal is
processed
+ // asynchronously on the DAGScheduler event loop.
+ val submitted = new ConcurrentLinkedQueue[Int]()
+ sc.addSparkListener(new SparkListener {
+ override def onStageSubmitted(e: SparkListenerStageSubmitted): Unit =
+ submitted.add(e.stageInfo.stageId)
+ })
+ assert(shuffled.collect().length === 8)
+ sc.listenerBus.waitUntilEmpty()
+ assert(submitted.size() === 1, s"expected only the reduce stage, got
$submitted")
+ assert(tracker.getNumAvailableOutputs(shuffleId) === 4)
+ }
+ }
+
+ test("SPARK-58828: holdExecutors drains the executors and resumeExecutors
brings them back") {
+ verifyHoldAndResumeExecutors(
+ new SparkConf().setAppName("test").setMaster("local-cluster[1,1,1024]")
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(DECOMMISSION_ENABLED, true))
+ }
+
+ test("SPARK-58828: hold and resume the executors with dynamic allocation") {
+ verifyHoldAndResumeExecutors(
+ new SparkConf().setAppName("test").setMaster("local-cluster[1,1,1024]")
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(DECOMMISSION_ENABLED, true)
+ .set(DYN_ALLOCATION_ENABLED, true)
+ .set(DYN_ALLOCATION_INITIAL_EXECUTORS, 1)
+ .set(DYN_ALLOCATION_MIN_EXECUTORS, 1))
+ }
+
+ test("SPARK-58828: a task running at the hold finishes and a pending one
runs after resume") {
+ val conf = new
SparkConf().setAppName("test").setMaster("local-cluster[1,1,1024]")
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(DECOMMISSION_ENABLED, true)
+ withExternalShuffleServer(conf) {
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 1, 60000)
+ val taskStarted = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit =
taskStarted.release()
+ })
+ // Two tasks on one core: the second is still pending when the hold
starts
+ val result = sc.parallelize(1 to 2, 2).map { i => Thread.sleep(2000); i
* 10 }.collectAsync()
+ assert(taskStarted.tryAcquire(1, 60, TimeUnit.SECONDS))
+ assert(sc.holdExecutors())
+ // The running task finishes before the executor exits
+ eventually(timeout(60.seconds)) {
+ assert(sc.getExecutorIds().isEmpty)
+ }
+ assert(sc.resumeExecutors())
+ // The pending task runs after the resume and the job completes with no
lost work
+ assert(ThreadUtils.awaitResult(result, 2.minutes).sorted === Seq(10, 20))
+ }
+ }
+
+ test("SPARK-58828: restoring the hold invariant pushes a zero requirement
and drains") {
+ sc = new SparkContext(new
SparkConf().setAppName("test").setMaster("local"))
+ val backend = mock(classOf[CoarseGrainedSchedulerBackend])
+ when(backend.republishRequestedTotals()).thenReturn(true)
+ when(backend.getExecutorIds()).thenReturn(Seq("1", "2"))
+ when(backend.decommissionExecutors(any(), any(),
any())).thenReturn(Seq("1", "2"))
+ assert(sc.zeroExecutorRequirementAndDrain(backend))
+ verify(backend).republishRequestedTotals()
+ verify(backend).decommissionExecutors(any(), meq(false), meq(false))
+
+ // A failed publish must not abort the drain
+ val failing = mock(classOf[CoarseGrainedSchedulerBackend])
+ when(failing.republishRequestedTotals()).thenThrow(new
RuntimeException("boom"))
+ when(failing.getExecutorIds()).thenReturn(Seq("3"))
+ when(failing.decommissionExecutors(any(), any(),
any())).thenReturn(Seq("3"))
+ assert(!sc.zeroExecutorRequirementAndDrain(failing))
+ verify(failing).decommissionExecutors(any(), meq(false), meq(false))
+ }
}
object SparkContextSuite {
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
index 6d01593f3e1a..ecdd02585a12 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -793,6 +793,96 @@ class CoarseGrainedSchedulerBackendSuite extends
SparkFunSuite with LocalSparkCo
assert(infos.last.requestTime.isEmpty)
}
+ test("SPARK-58828: New registered executor should be decommissioned while
held") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val mockEndpointRef = new MockExecutorRpcEndpointRef(conf)
+ val mockAddress = mock[RpcAddress]
+
+ backend.setExecutorsHeld(true)
+ backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map(), Map(),
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+
+ sc.listenerBus.waitUntilEmpty(executorUpTimeout.toMillis)
+ assert(mockEndpointRef.decommissionReceived)
+ // The zero requirement is re-published without touching the requested
totals
+ assert(backend.getRequestedTotalExecutors().isEmpty)
+ }
+
+ test("SPARK-58828: reset clears the explicit request record with the
requested totals") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val defaultProf = sc.resourceProfileManager.defaultResourceProfile
+ sc.requestTotalExecutors(2, 0, Map.empty)
+ assert(backend.hasExplicitExecutorRequests)
+
+ // While held, reset() preserves the requested totals and the explicit
record
+ backend.setExecutorsHeld(true)
+ backend.reset()
+ assert(backend.hasExplicitExecutorRequests)
+ assert(backend.getRequestedTotalExecutors().getOrElse(defaultProf, -1) ===
2)
+
+ // Not held, reset() clears both
+ backend.setExecutorsHeld(false)
+ backend.reset()
+ assert(!backend.hasExplicitExecutorRequests)
+ assert(backend.getRequestedTotalExecutors().isEmpty)
+ }
+
+ test("SPARK-58828: the restore sentinel is published without being
recorded") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val defaultProf = sc.resourceProfileManager.defaultResourceProfile
+ assert(backend.publishTotalsWithoutRecording(Map(defaultProf ->
Int.MaxValue)))
+ // Neither the totals nor the explicit-request record change, so
killExecutors' empty-map
+ // seeding and a later hold's kill-seeded restore keep their pre-hold
behavior
+ assert(backend.getRequestedTotalExecutors().isEmpty)
+ assert(!backend.hasExplicitExecutorRequests)
+ }
+
+ test("SPARK-58828: executor requests made while held are retained until
resume") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val defaultProf = sc.resourceProfileManager.defaultResourceProfile
+ assert(!backend.hasExplicitExecutorRequests)
+
+ backend.setExecutorsHeld(true)
+ // Both public request APIs keep recording the requested totals while held
+ sc.requestTotalExecutors(3, 0, Map.empty)
+ assert(backend.getRequestedTotalExecutors().getOrElse(defaultProf, -1) ===
3)
+ sc.requestExecutors(2)
+ assert(backend.getRequestedTotalExecutors().getOrElse(defaultProf, -1) ===
5)
+ assert(backend.hasExplicitExecutorRequests)
+
+ // A held reassertion publishes zero but does not overwrite the requested
totals
+ backend.reassertHeldRequirement()
+ assert(backend.getRequestedTotalExecutors().getOrElse(defaultProf, -1) ===
5)
+
+ // Resume republishes the requested totals as-is, and a stale reassertion
after the hold
+ // is lifted leaves them alone
+ backend.setExecutorsHeld(false)
+ assert(backend.republishRequestedTotals())
+ backend.reassertHeldRequirement()
+ assert(backend.getRequestedTotalExecutors().getOrElse(defaultProf, -1) ===
5)
+ }
+
test("UpdateUserCredentials is broadcast to all registered executors") {
val conf = new SparkConf()
.setMaster("local-cluster[0, 3, 1024]")
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index f4dcdaf96d7c..a44ed77b6511 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -395,6 +395,11 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
override protected def outstandingTasksForOtherWork(rpId: Int,
excludeStageIds: Set[Int]): Int =
outstandingTasksForOtherWorkForTest(rpId, excludeStageIds)
+ // Seam for the hold state (SPARK-58828) read by the barrier retry-budget
freeze. Default
+ // false so existing tests are unaffected.
+ @volatile var executorsHeldForTest: Boolean = false
+ override protected def executorsHeld: Boolean = executorsHeldForTest
+
/**
* Schedules shuffle merge finalize.
*/
@@ -6868,6 +6873,27 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
}
}
+ test("SPARK-58828: a barrier job does not consume its retry budget while
held") {
+ // 4 barrier tasks on a local[2] backend fail the slot check; while held
that must not
+ // count against
spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures.
+ val barrierRdd = new MyRDD(sc, 4, Nil).barrier().mapPartitions(iter =>
iter)
+ val myScheduler = scheduler.asInstanceOf[MyDAGScheduler]
+ myScheduler.executorsHeldForTest = true
+ try {
+ val failure = new
java.util.concurrent.atomic.AtomicReference[Exception]()
+ val failListener = new JobListener {
+ override def taskSucceeded(index: Int, result: Any): Unit = {}
+ override def jobFailed(exception: Exception): Unit =
failure.set(exception)
+ }
+ submit(barrierRdd, Array(0, 1, 2, 3), listener = failListener)
+ assert(failure.get() === null, "a barrier job submitted while held must
wait, not fail")
+ assert(scheduler.barrierJobIdToNumTasksCheckFailures.isEmpty,
+ "the retry budget must not be consumed while held")
+ } finally {
+ myScheduler.executorsHeldForTest = false
+ }
+ }
+
test("pipelined shuffle: an over-capacity 3-stage all-pipelined chain is
rejected up front") {
// A 3-stage all-pipelined chain A->B->C has whole-group demand 2+2+2 = 6.
With capacity 3 it
// cannot co-fit, so the up-front gang admission check
(handleJobSubmitted) fails the job before
diff --git a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
index 9096076f8212..7fe24572cae1 100644
--- a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
@@ -302,6 +302,29 @@ class UISeleniumSuite extends SparkFunSuite with
WebBrowser with Matchers {
}
}
+ test("SPARK-58828: spark.ui.holdEnabled should properly control hold button
display") {
+ def hasHoldLink: Boolean = find(className("confirm-link")).isDefined
+ // The control needs a coarse-grained backend (so local-cluster, not
local) and the hold
+ // preconditions; the external shuffle service itself is not needed to
render the page.
+ val holdConfs = Map(
+ SHUFFLE_SERVICE_ENABLED.key -> "true",
+ DECOMMISSION_ENABLED.key -> "true")
+
+ withSpark(newSparkContext(master = "local-cluster[1,1,1024]",
+ additionalConfs = holdConfs)) { sc =>
+ eventually(timeout(10.seconds), interval(50.milliseconds)) {
+ goToUi(sc, "/jobs")
+ assert(hasHoldLink)
+ }
+ }
+
+ withSpark(newSparkContext(master = "local-cluster[1,1,1024]",
+ additionalConfs = holdConfs + (UI_HOLD_ENABLED.key -> "false"))) { sc
=>
+ goToUi(sc, "/jobs")
+ assert(!hasHoldLink)
+ }
+ }
+
test("jobs page should not display job group name unless some job was
submitted in a job group") {
withSpark(newSparkContext()) { sc =>
// If no job has been run in a job group, then "(Job Group)" should not
appear in the header
diff --git a/docs/configuration.md b/docs/configuration.md
index 38460273a823..616367592b62 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1605,6 +1605,21 @@ Apart from these, the following properties are also
available, and may be useful
</td>
<td>1.0.0</td>
</tr>
+<tr>
+ <td><code>spark.ui.holdEnabled</code></td>
+ <td>true</td>
+ <td>
+ Allows the whole application to be held and resumed from the web UI.
Holding gracefully
+ decommissions all executors and stops requesting new ones. Cached blocks
are not preserved
+ and are recomputed after resuming. This takes effect only when
+ <code>spark.decommission.enabled</code> is true, the shuffle data is kept
outside the
+ executors (through either <code>spark.shuffle.service.enabled</code> or a
+ <code>ShuffleDataIO</code> with reliable storage), and the cluster manager
can hold
+ executors: Standalone, YARN, and Kubernetes with
+ <code>spark.kubernetes.allocation.pods.allocator=direct</code>.
+ </td>
+ <td>4.4.0</td>
+</tr>
<tr>
<td><code>spark.ui.threadDumpsEnabled</code></td>
<td>true</td>
diff --git a/docs/web-ui.md b/docs/web-ui.md
index 6ae0a363d187..085e8be75537 100644
--- a/docs/web-ui.md
+++ b/docs/web-ui.md
@@ -70,6 +70,24 @@ The information displayed at the top of the page includes:
The current user, application start time, and total uptime are shown in the
footer at the
bottom of every page.
+When the application can be held, the summary shows an **Application** line
with a **(hold)**
+link; clicking it stops requesting new executors and gracefully decommissions
the running ones,
+so each finishes its tasks and then exits (unless
`spark.executor.decommission.forceKillTimeout`
+is set, which kills a still-busy executor after that timeout). The line then
reads `Held` with a
+**(resume)** link that restores the executor requirement. Shuffle output
written before the hold
+stays available, but cached blocks are recomputed after resuming, and an RDD
created while held
+is sized against a default parallelism of 2 (no executors are alive) and keeps
that partition
+count afterwards. Pipelined-shuffle jobs and workloads with long-running tasks
(streaming
+receivers, continuous processing) are outside the hold's scope: a hold
requested while a
+pipelined job is running is rejected, a pipelined job submitted while held
fails immediately
+and should be resubmitted after the resume (unless the internal
+`spark.scheduler.pipelinedGroup.slotCheck.enabled` is false, in which case
there is no
+admission check and it waits for the resume), and a long-running task never
finishes, so the
+drain cannot complete. The control only appears when `spark.ui.holdEnabled` is
true,
+`spark.decommission.enabled` is true, the shuffle data is kept outside the
executors, and the
+cluster manager can hold executors (Standalone, YARN, and Kubernetes with the
`direct` pods
+allocator); see [Configuration](configuration.html#spark-ui).
+
<p style="text-align: center;">
<img src="img/AllJobsPage.png" title="All Jobs page" alt="All Jobs page"
width="100%"/>
</p>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
