peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3815331354
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2093,8 +2093,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 (sc.executorsHeld) {
Review Comment:
**Finding 29.** This fix identified the right class — a consumer that reads
zero slots must wait for the resume, not fail — but there is a second consumer
of the same count, and it is terminal. Anchoring here because that's where the
pattern lives; the code in question is `rejectUnadmittablePipelinedGroup` at
`core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:1844`.
`rejectUnadmittablePipelinedGroup` computes `totalSlots =
maxConcurrentTasksForProfile(rp.id)` → `sc.maxNumConcurrentTasks` →
`CoarseGrainedSchedulerBackend.maxNumConcurrentTasks`, which filters on
`isExecutorActive`. `drainHeldExecutors` puts every executor into
`executorsPendingDecommission`, so that count is 0 from the moment the hold
starts. Your own scaladoc there is explicit that the consequence is terminal:
"pipelined admission is TERMINAL (one check, then fail), delegating
transient-shortfall retry to the caller."
`spark.scheduler.pipelinedGroup.slotCheck.enabled` defaults to `true`.
I measured both halves against `eb38c362`.
Capacity, on a real `local-cluster[1,1,1024]` with the shuffle service and
decommissioning enabled, calling `holdExecutors()` for real:
```
slots-before=1 slots-after-hold=0 executors-still-registered=1
```
So it is 0 as soon as `holdExecutors()` returns — this does not wait for the
drain to finish.
The check's response to 0, via `DAGSchedulerSuite`'s existing
`maxConcurrentTasksForTest` seam:
```
failed-immediately=true stages-submitted=0
msg=[CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT] Cannot run the pipelined stage
group: it needs 4
concurrent task slots to run all its stages together, but only 0 are
currently free. ...
```
Failed outright, no stage created, no retry. Since the comparison is `demand
> freeSlots` against `freeSlots = 0`, this holds for any pipelined group size —
the 4 above is just the group I used. The one step I did not measure end to end
is the seam substitution itself, i.e. that the value the production
`maxConcurrentTasksForProfile` reads is the same 0 the first measurement shows;
that is a one-line body.
The config combination is reachable: `rejectUnsupportedPipelinedJob`
(`:1156-1174`) rejects pipelined shuffle under speculation or dynamic
allocation, so pipelined implies DA off — precisely the
`executorAllocationManager.isEmpty` path this PR spent three rounds on. Add
`spark.decommission.enabled` and an external shuffle service and you have a
supported hold. A real-time-mode streaming query held between micro-batches, so
`hasPipelinedTaskSets` is false at that instant, then dies on the next batch
instead of pausing.
The fix is the mechanism you already have here — re-post rather than fail
while held:
```scala
if (demand > freeSlots) {
if (sc.executorsHeld) {
// The executors are held, so there are no slots by construction.
Wait for the resume
// like a barrier job rather than failing: the shortfall is not
transient scarcity.
logInfo(log"Deferring job ${MDC(JOB_ID, jobId)}: the executors are
held")
messageScheduler.schedule(new Runnable {
override def run(): Unit = eventProcessLoop.post(JobSubmitted(...))
}, timeIntervalNumTasksCheck, TimeUnit.SECONDS)
return true
}
...
```
`DAGSchedulerSuite` already has four tests asserting
`CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT` on this path, so one more asserting
the job is *not* failed while held is a small addition — it needs the same seam
as finding 26.
If you'd rather keep this out of scope, the honest alternative is to say so
in `docs/web-ui.md`: right now it says a hold is rejected while a pipelined job
is running, which reads as though the interaction is handled in both directions.
##########
core/src/test/scala/org/apache/spark/SparkContextSuite.scala:
##########
@@ -1578,6 +1584,97 @@ 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 verifyHoldAndResumeExecutors(conf: SparkConf): 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)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 1, 60000)
+ assert(sc.executorHoldSupported)
+ assert(!sc.executorsHeld)
+
+ assert(sc.holdExecutors())
Review Comment:
**Finding 19.** No test in this PR has a task running when the hold happens,
and none produces shuffle output across one — so neither of the two guarantees
`holdExecutors()`'s scaladoc makes is exercised.
`verifyHoldAndResumeExecutors` never submits a job: it waits for the
executor, holds, waits for the drain (now with the added settle check),
resumes, waits for a new executor. `ExecutorAllocationManagerSuite` drives
`suspend()`/`resume()` against a mocked client; the
`CoarseGrainedSchedulerBackendSuite` cases register a mock endpoint or poke the
backend directly. No tasks anywhere. So "each executor finishes its running
tasks and then exits, while the shuffle data already written remains available
outside the executors" is asserted nowhere — and that sentence is why
`executorHoldSupported` demands ESS-or-reliable-storage. If that gate, or
`ExecutorDecommissionInfo`'s `workerHost = None`, ever drifted,
`handleExecutorLost` would flip `fileLost` to true
(`core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:4105`),
every map output would be unregistered on the drain, and the only symptom would
be a stage silently recomputed after resume.
I wrote the assertions below and ran them against `eb38c362` inside the
existing helper — they pass, in 5.3 s:
```
map-outputs-before=4 map-outputs-after=4
stages-submitted-by-second-action=1 ids=[3]
```
So the guarantee does hold today; the ask is to pin it, not to fix it. The
single stage id is the part that matters — the map stage really is skipped
rather than recomputed, so the oracle discriminates.
```scala
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.resumeExecutors())
...
// The map output written before the hold survived the drain: only the
reduce stage re-runs
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)
```
(`SparkListenerStageSubmitted` needs adding to the existing
`org.apache.spark.scheduler` import.) Assert the map output *after* the second
job rather than right after the drain: `handleExecutorLost` runs asynchronously
on the DAGScheduler event loop, so an assertion straight after
`getExecutorIds().isEmpty` can pass before the removal has been processed.
For the running-tasks half, `WorkerDecommissionSuite`'s `test("verify a
running task with all workers decommissioned succeeds")` is the shape — a slow
job started before the hold, asserted to complete. That one also happens to be
the only thing that would catch finding 32.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,213 @@ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * The 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.
+ *
+ * @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 {
Review Comment:
**Finding 30.** Two things about this gate: it can be defeated, and the
scaladoc's "best-effort" undersells what losing the race costs.
**The window is wider than it looks.** `hasPipelinedTaskSets` reads
`taskSetsByStageIdAndAttempt`, so it only sees groups whose task sets have
already reached the `TaskSchedulerImpl`. A pipelined job that has passed
`submitJob` but whose stages are still being created on the DAGScheduler event
loop is invisible here, and the check is evaluated *outside* `synchronized`, so
`submitTasks` can also land between the check and `drainHeldExecutors`.
**What losing it costs.** `TaskSetManager.effectiveMaxTaskFailures` is `if
(taskSet.isPipelined) 1 else maxTaskFailures`
(`core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:80`), so
a single task lost to the decommission aborts the whole group — not a slow
drain, a failed user job. Worth saying that in the scaladoc even if the race
stays.
Two fixes, and they close different halves:
- For the narrow window, move the check next to the drain under one
acquisition of the scheduler monitor — `decommissionExecutors` already takes it
via `withLock`, so a `TaskSchedulerImpl` helper along the lines of `def
decommissionAllUnlessPipelined(...): Option[Seq[String]] = synchronized { if
(hasPipelinedTaskSets) None else Some(...) }` makes it atomic. Do **not** widen
that lock to cover `cg.republishRequestedTotals()`, which blocks in
`awaitResult` for up to `spark.rpc.askTimeout`.
- For the wide one, ask the DAGScheduler instead of the TaskScheduler:
`jobIdToActiveJob.values.exists(_.hasPipelinedDependency)` is the authoritative
"a pipelined job is live" set and is populated at job submission
(`DAGScheduler.scala:2148`), so it covers the pre-taskset window that
`hasPipelinedTaskSets` cannot.
If you fix finding 29 as suggested, the submit-side gate closes both halves
on its own and this check becomes belt-and-braces.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2093,8 +2093,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 (sc.executorsHeld) {
Review Comment:
**Finding 26.** Three branches were added across this PR to satisfy review
findings, and none of them has a test. This one is also untestable as written,
which is the part worth fixing first.
`sc.executorsHeld` can only become true through `holdExecutors()`, which
needs a `CoarseGrainedSchedulerBackend`. `DAGSchedulerSuite` and
`BarrierStageOnSubmittedSuite` both run on `local[…]`, whose
`LocalSchedulerBackend` isn't one, so `sc.executorsHeld` is unreachably `false`
there and no assertion can distinguish this branch from the old code.
This file already solved exactly this problem, one screen up, and documents
why: `protected def maxConcurrentTasksForProfile` and
`outstandingTasksForOtherWork` are "Extracted as a seam so tests can control it
without changing the cluster's core count". The same idiom applies:
```scala
/** Extracted as a seam so tests can control the hold state. */
protected def executorsHeld: Boolean = sc.executorsHeld
```
read here and in `rejectUnadmittablePipelinedGroup`, after which
`BarrierStageOnSubmittedSuite`'s "requires more slots than the total number of
slots" tests are the template for asserting the retry budget is not consumed —
and finding 29 becomes testable by the same seam. (I used the two existing
seams to measure finding 29, which is how I know the pattern carries.)
The other two are cheap against existing precedents:
- `KubernetesClusterSchedulerBackend.supportsExecutorHold` — the one place
the guard can be wrong per allocator value — has no test.
`KubernetesClusterSchedulerBackendSuite` already builds a backend from a
`SparkConf`; three lines per allocator value.
- The UI gate has none either, and `UISeleniumSuite` has both halves ready:
`"spark.ui.killEnabled should properly control kill button display"` for the
link and the `/jobs/job/kill/?id=0` GET-response test for the endpoint. Its
`newSparkContext` takes a `killEnabled` flag, so a `holdEnabled` twin is
mechanical — with `local-cluster`, since the control doesn't render under
`local`.
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +66,85 @@ 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, rendered by
AllJobsPage next
+ // to the Application line: Some(message) while a request is running or
after it was not
+ // acknowledged, None when idle or after a success.
+ @volatile private var holdRequestStatus: Option[String] = None
+
+ private[jobs] def lastHoldRequestStatus: Option[String] = holdRequestStatus
+
+ def handleHoldRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.foreach { pool =>
+ holdRequestStatus = Some("hold requested")
Review Comment:
**Finding 28.** Two things, both visible on the page.
One field carries both operations, and a failure message is only cleared by
a later *success* — so "the last hold request was not acknowledged, see the
driver logs" renders next to the **(resume)** link after the hold has in fact
taken effect, and survives every page load until someone resumes successfully.
Tagging it fixes both:
```scala
// (isHold, message); AllJobsPage renders it only next to the matching
control
@volatile private var holdRequestStatus: Option[(Boolean, String)] = None
```
Second, "not acknowledged" is now wrong for two of the three ways
`holdExecutors()` returns `false`. The pipelined rejection
(`SparkContext.scala:2141`) and the unsupported-backend branch never ask the
cluster manager at all, and with dynamic allocation the scaladoc says a
rejected push "is retried in the background; the executors are drained in
either case" — so the one case where the message is accurate is the static
rejected publish. Distinguishing them needs more than a boolean out of
`holdExecutors()`; the cheap version is to soften the wording to "the last hold
request did not take effect, see the driver logs", and the better one is to
have `holdExecutors()` report *why* it declined so the page can say "a
pipelined job is running".
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,213 @@ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
Review Comment:
**Finding 23.** This scaladoc is now the accurate description of the feature
and the PR body is three commits behind it. Four specifics, all of which a
reader would act on:
1. "Holding stops requesting new executors (via
`ExecutorAllocationManager.suspend()` when dynamic allocation is on, or
**`requestTotalExecutors(0)`** otherwise)" — it no longer does that, and the
difference is the point of `8bfca77`: `publishTotals()` publishes zeros
*without recording them*, which is what lets resume restore the pre-hold
requirement.
2. The paragraph added here — requirements requested during a hold are
retained and applied on resume — is new user-visible behavior for
`sc.requestExecutors` / `sc.requestTotalExecutors` and isn't mentioned.
3. The pipelined rejection isn't mentioned (it is in `web-ui.md` and in the
scaladoc two paragraphs down).
4. Barrier jobs no longer consume their retry budget while held — a
`DAGScheduler` behavior change that isn't in the body either, and the
description's file list gives no hint that `DAGScheduler` and
`TaskSchedulerImpl` are touched at all.
This is the third round where a fix changed the contract and the body didn't
follow (11 and 17 were the same shape), so a pass over the whole description
before merge is probably cheaper than another round of these.
##########
docs/web-ui.md:
##########
@@ -70,6 +70,16 @@ 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. The line then reads `Held` with a
**(resume)** link
Review Comment:
**Finding 32.** This sentence has an unnamed precondition.
`drainHeldExecutors` → `decommissionExecutors` reaches the force-kill block at
`core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:634`,
so when `spark.executor.decommission.forceKillTimeout` is set, executors still
running tasks are killed `cleanupInterval` seconds into the hold —
`killExecutors(stragglers, false, false, force = true)`. That directly
contradicts three pieces of prose:
- this line: "so each finishes its tasks and then exits"
- the `holdExecutors()` scaladoc: "Each executor finishes its running tasks
and then exits"
- the PR description: "Executors are drained gracefully with no loss of
in-progress task work"
The config has no default, so most clusters are fine — but it exists
precisely for operators who don't want unbounded drains, which is the same
audience that wants a hold. On those clusters a hold silently loses in-progress
task work, and for a pipelined group it aborts the whole group
(`effectiveMaxTaskFailures = 1`, see finding 30).
Either name it here and in the scaladoc — "unless
`spark.executor.decommission.forceKillTimeout` is set, in which case executors
still running tasks are killed after that timeout" — or give
`decommissionExecutors` a way to skip the force-kill schedule for a hold-driven
drain, since a hold has no deadline of its own to protect.
##########
core/src/main/scala/org/apache/spark/internal/config/UI.scala:
##########
@@ -92,6 +92,18 @@ 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 " +
Review Comment:
**Finding 20.** `executorHoldSupported` is a conjunction of three things;
every user-facing description of it names two.
The missing one is the backend: a `CoarseGrainedSchedulerBackend` whose
`supportsExecutorHold` is true. So a user on `local[*]`, or on Kubernetes with
`spark.kubernetes.allocation.pods.allocator` set to `statefulset`, `deployment`
or a custom class, sets both documented configs, sees no `(hold)` line at all,
and has nothing to tell them why. Four sites say it the same incomplete way —
here, `docs/configuration.md:1614`, `docs/web-ui.md:79` and the PR description.
The `holdExecutors()` and `executorHoldSupported` scaladocs *do* get it right,
which is what makes the omission a consistent one rather than a slip. Appending
something like "…and a cluster manager that can hold executors (not local mode;
on Kubernetes only with `spark.kubernetes.allocation.pods.allocator=direct`)"
to each covers it.
Second, smaller point at the same anchor: the two unmet-precondition paths
behave differently and neither is documented. `holdExecutors()` now matches
`case cg: CoarseGrainedSchedulerBackend if cg.supportsExecutorHold`, so an
unsupported *backend* falls to `case _` and returns `false` with a warning,
while an unmet `spark.decommission.enabled` or shuffle-storage condition hits
`require(...)` and throws `IllegalArgumentException`. `SparkContextSuite:1595`
pins the throw, so it's deliberate — and `killExecutors` uses `require` too —
but the scaladoc promises only `@return whether the lowered executor
requirement was acknowledged`, and the description's "It is rejected otherwise"
reads as a `false` return. An `@throws` clause would settle it.
Worth mentioning while you're in this doc: it's the natural place to record
that fallback storage (`spark.storage.decommission.fallbackStorage.path`)
deliberately does *not* qualify. It does migrate shuffle blocks, so a user with
it set will expect to be covered, but `handleExecutorLost` computes `fileLost =
true` for that configuration, so anything not yet migrated when the executor
exits is dropped — a weaker guarantee than the scaladoc's, and a good reason to
exclude it. Saying so beats leaving it implicit.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -739,6 +778,95 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
executorDataMap.keySet.toSeq
}
+ /** See `SparkContext.holdExecutors()`. */
+ private[spark] def setExecutorsHeld(held: Boolean): Unit = {
+ executorsHeld = held
+ }
+
+ /**
+ * Whether this backend can hold the executors gracefully. Overridden by
backends whose
+ * cluster manager would terminate running executors on a zero requirement.
+ */
+ private[spark] def supportsExecutorHold: Boolean = true
Review Comment:
**Finding 31.** The default makes an unknown backend opt *in*, and the
scaladoc scopes the exception too narrowly — "backends whose cluster manager
would terminate running executors on a zero requirement". There is a second way
to be unable to hold: not being able to honour a total at all.
`doRequestTotalExecutors` in this class returns `Future.successful(false)`
(`:1050`). A backend supplied through the `ExternalClusterManager` SPI that
extends `CoarseGrainedSchedulerBackend` without overriding it inherits
`supportsExecutorHold = true`. With `spark.decommission.enabled` and an
external shuffle service on — and `spark.ui.holdEnabled` defaulting to `true` —
the `(hold)` link renders. The click then *works*, because `drainHeldExecutors`
goes through `decommissionExecutors`, which never touches
`doRequestTotalExecutors`. `resumeExecutors()` afterwards gets `false` from
every publish, so it takes the failure path, re-arms the guard and re-drains —
forever. The application is held with no way out short of a restart.
Nothing in the tree hits this: all three subclasses override
`doRequestTotalExecutors`. It's the polarity of a new extension point that I'd
change, so the default is safe rather than convenient:
```scala
private[spark] def supportsExecutorHold: Boolean = false
```
with `override def supportsExecutorHold: Boolean = true` in
`StandaloneSchedulerBackend` and `YarnSchedulerBackend`, and the existing
allocator check kept in `KubernetesClusterSchedulerBackend`. Honest
counter-argument: three overrides instead of one, and the current denylist is
correct for everything in-tree. But the K8s comment in the same commit already
reaches for opt-in reasoning for pod allocators ("and unknown custom ones"), so
the two guards would end up consistent — and the failure mode here is worse
than the one the K8s guard prevents, because it isn't recoverable in-process.
##########
docs/web-ui.md:
##########
@@ -70,6 +70,16 @@ 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. 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. A hold requested while a
pipelined job is
Review Comment:
**Finding 25.** One more consequence of a hold that outlives it, and it's
the kind users will misdiagnose.
`CoarseGrainedSchedulerBackend.defaultParallelism()` is
`conf.getInt(spark.default.parallelism, math.max(totalCoreCount.get(), 2))`,
and `totalCoreCount` is decremented as each executor is removed (`:515`). Once
the drain finishes it is 0, so with `spark.default.parallelism` unset — the
default — `sc.defaultParallelism` is **2** for the duration of the hold.
That is not transient. Any RDD created while held — `sc.parallelize(seq)`,
`sc.textFile(path)`, a shuffle sized from `defaultParallelism` — is fixed at 2
partitions and keeps it after resume. A driver that keeps building work during
a hold comes back with permanently under-partitioned RDDs, no error, and
nothing pointing at the hold.
Dynamic-allocation users scaled to zero already have this, so the mechanism
is pre-existing; what's new is that any operator can trigger it deliberately,
from a button. A sentence next to the cached-blocks one is enough: "While held
the application has no executors, so `spark.default.parallelism` falls back to
2; RDDs created during a hold keep that partition count after resuming."
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,213 @@ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * The 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.
+ *
+ * @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
Review Comment:
**Finding 27.** The sentinel leaks into the request-time bookkeeping and
never leaves.
On a standalone resume, `requestTotalExecutors(Map(default ->
Int.MaxValue))` runs `updateExecRequestTimes`, whose delta is `Int.MaxValue -
0`, so `updateExecRequestTime` pushes `(Int.MaxValue, now)` onto
`execRequestTimes`. The `RegisterExecutor` handler dequeues the head and
re-prepends `(count - 1, sameTs)` while `count > 1`
(`CoarseGrainedSchedulerBackend.scala:298-312`), so that one entry outlives
every executor: from then on each registration records the resume timestamp as
its request time, which reaches the event log as `"Request Time"`
(`core/src/main/scala/org/apache/spark/util/JsonProtocol.scala:815`).
Only history data is affected, hence Minor. Simplest is to skip the
bookkeeping for the unbounded sentinel; tidier is to stop routing
`Int.MaxValue` through `requestTotalExecutors` at all — on standalone "restore
the unbounded default" is really "forget the requirement", which
`requestedTotalExecutorsPerResourceProfile.clear()` plus a publish says without
a magic number, and would make the saturating add at `:946` defensive rather
than required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
