peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3805557314
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -282,12 +293,75 @@ 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.
+ suspendedSyncPending = true
+ }
+ }
+
+ /**
+ * 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.
`schedule()` is a
+ * no-op until [[resume()]] is called.
+ */
+ def suspend(): Unit = synchronized {
+ if (!suspended) {
+ suspended = true
+ numExecutorsTargetPerResourceProfileId.keys.foreach { rpId =>
+ numExecutorsTargetPerResourceProfileId(rpId) = 0
+ }
+ numExecutorsToAddPerResourceProfileId.keys.foreach { rpId =>
+ numExecutorsToAddPerResourceProfileId(rpId) = 1
+ }
+ if (!testing) {
Review Comment:
**Finding 2.** The acknowledgment is discarded here (and in `resume()`), so
a request the cluster manager rejects is never re-sent.
`client.requestTotalExecutors` returns `false` on a real, recoverable path:
`YarnSchedulerEndpoint` replies `false` whenever `amEndpoint` is empty
("Attempted to request executors before the AM has registered"). Everywhere
else in this file the ack is checked -- `doUpdateRequest` rolls the target back
on `false` so the next tick retries, and you added `suspendedSyncPending` to
get exactly that retry for `reset()`. `suspend()` is the one push with neither:
after a `false` the driver believes the target is 0, `schedule()` returns early
on every tick without re-pushing, and the cluster manager keeps granting
executors against its stale target -- each registers and is immediately
decommissioned by the new `executorsHeld` branch, so the application churns
executors for as long as it is held.
The retry you already built covers it, it just needs arming from here:
```scala
if (!(testing || client.requestTotalExecutors(
numExecutorsTargetPerResourceProfileId.toMap,
numLocalityAwareTasksPerResourceProfileId.toMap,
rpIdToHostToLocalTaskCount))) {
suspendedSyncPending = true
}
```
`resume()` has the mirror hole: if its push is rejected, `suspended` is
cleared anyway, and `updateAndSyncNumExecutorsTarget` only calls the client
when a target actually *changes*. With `spark.dynamicAllocation.minExecutors =
1` and no pending tasks the target settles back on 1, `delta == 0`, no request
is sent, and the application stays at zero executors on the cluster-manager
side until some later backlog moves the target.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
+ *
+ * This requires both an external shuffle service
(`spark.shuffle.service.enabled`) and
+ * decommission support (`spark.decommission.enabled`).
+ *
+ * @return whether the request is received by the cluster manager.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires both ${SHUFFLE_SERVICE_ENABLED.key} and "
+
+ s"${DECOMMISSION_ENABLED.key}")
+ synchronized {
+ if (!_executorsHeld) {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume. Standalone ignores
+ // spark.executor.instances and has no explicit executor
requirement by default,
+ // so restore an unbounded one, even when the conf carries a
leftover value.
+ // Elsewhere follow the conf, or fall back to the cluster
manager's default when
+ // no executor has registered yet.
+ heldNumExecutors = schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.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
+ b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.setExecutorsHeld(true)
+ case _ =>
+ }
+ executorAllocationManager match {
+ case Some(manager) => manager.suspend()
+ case None =>
+ b.requestTotalExecutors(
+
immutable.Map(resourceProfileManager.defaultResourceProfile.id -> 0),
Review Comment:
**Finding 4.** With dynamic allocation off nothing re-asserts this zero
target after the cluster manager loses the driver's state, so the hold is
silently lost on the requirement side.
`YarnSchedulerBackend.reset()` (YARN client mode, AM re-registration) runs
`super.reset()`, which clears `requestedTotalExecutorsPerResourceProfile`, and
then `sc.executorAllocationManager.foreach(_.reset())` -- the `foreach` is a
no-op here, so the `suspendedSyncPending` re-push you added never happens on
this path. The restarted AM initialises its target from
`SchedulerBackendUtils.getInitialTargetExecutorNumber(conf)`, i.e.
`spark.executor.instances`, and starts allocating again; the new
`executorsHeld` branch then decommissions each one on registration, so the
application churns executors instead of staying held.
`CoarseGrainedSchedulerBackend` already knows it is held, so `reset()` is
the natural place -- and it can re-assert without the self-deadlock you
documented, as long as it does not await:
```scala
protected[scheduler] def reset(): Unit = {
val executors: Set[String] = synchronized {
requestedTotalExecutorsPerResourceProfile.clear()
if (executorsHeld) {
val defaultProf =
scheduler.sc.resourceProfileManager.defaultResourceProfile
requestedTotalExecutorsPerResourceProfile(defaultProf) = 0
doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
}
executorDataMap.keys.toSet
}
```
That would also make the dynamic-allocation path symmetric instead of
special-cased.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
+ *
+ * This requires both an external shuffle service
(`spark.shuffle.service.enabled`) and
+ * decommission support (`spark.decommission.enabled`).
+ *
+ * @return whether the request is received by the cluster manager.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires both ${SHUFFLE_SERVICE_ENABLED.key} and "
+
+ s"${DECOMMISSION_ENABLED.key}")
+ synchronized {
+ if (!_executorsHeld) {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume. Standalone ignores
+ // spark.executor.instances and has no explicit executor
requirement by default,
+ // so restore an unbounded one, even when the conf carries a
leftover value.
+ // Elsewhere follow the conf, or fall back to the cluster
manager's default when
+ // no executor has registered yet.
+ heldNumExecutors = schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.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
+ b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.setExecutorsHeld(true)
+ case _ =>
+ }
+ executorAllocationManager match {
Review Comment:
**Finding 1.** Both branches here can throw, and neither is caught, so
`holdExecutors()` can leave the application marked held with every executor
still running.
`manager.suspend()` and the `case None` branch both end up in
`CoarseGrainedSchedulerBackend.requestTotalExecutors`, which finishes with
`defaultAskTimeout.awaitResult(response)` -- that throws on an ask timeout or
an AM failure. `ExecutorAllocationManager.doUpdateRequest` wraps the very same
call in `catch { case NonFatal(e) => ... }` with the comment "Errors here are
more commonly caused by YARN AM restarts, which is a recoverable issue", so
this is a path that fires in practice.
When it throws, `_executorsHeld` and `setExecutorsHeld(true)` are already
set, but `b.decommissionExecutors(...)` below never runs. Nothing marked those
executors pending-decommission, so `isExecutorActive` stays true and they keep
being offered new tasks, while the Jobs page reports `Held (draining N
executors)` indefinitely. The comment two lines above says marking the hold
first avoids leaving it "half-held" -- but this *is* the half-held state: held
on the driver, untouched on the cluster.
Make the drain unconditional, e.g.
```scala
Utils.tryLogNonFatalError {
executorAllocationManager match {
case Some(manager) => manager.suspend()
case None =>
b.requestTotalExecutors(
immutable.Map(resourceProfileManager.defaultResourceProfile.id -> 0),
immutable.Map.empty, immutable.Map.empty)
}
}
val executors = b.getExecutorIds()
```
combined with the retry in finding 2, so a transient cluster-manager error
costs a re-push rather than the whole hold.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -321,6 +326,15 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
decommissionExecutors(Array((executorId, v._1)), v._2, v._3)
unknownExecutorsPendingDecommission.invalidate(executorId)
})
+ if (executorsHeld) {
Review Comment:
**Finding 3.** This branch has no test. It is a new execution-time path
inside an RPC handler, and its failure mode -- an executor the cluster manager
granted before the hold quietly surviving it -- raises no exception, so nothing
in CI would catch a regression.
`CoarseGrainedSchedulerBackendSuite` already has the shape ~250 lines above,
for the sibling `unknownExecutorsPendingDecommission` branch:
`test("SPARK-41766: New registered executor should receive decommission request
sent before registration")`. The held version is that test with one line
changed:
```scala
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)
}
```
(`decommissionExecutors` does not gate on `spark.decommission.enabled`,
which is why the SPARK-41766 test works without it.)
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -321,6 +326,15 @@ 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.
+ val decommissioned = decommissionExecutors(
+ Array((executorId, ExecutorDecommissionInfo("Executors are
held"))),
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ scheduler.sc.executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
Review Comment:
**Finding 9.** This call is a no-op on this path. `ExecutorMonitor` is
registered with `listenerBus.addToManagementQueue(...)`, so it builds its
tracker asynchronously from the `SparkListenerExecutorAdded` posted ~15 lines
above; by the time `executorsDecommissioned` runs here, `executors.get(id)` is
still `null` and the `decommissioning = true` assignment is dropped. So the
executor never shows up in `numberDecommissioningExecutors` or
`decommissioningPerResourceProfileId` -- the two things the comment above says
the call is for.
Harmless while held, since `schedule()` returns early, but it means the
metric under-reports exactly the executors the hold had to drain. Either drop
the call here (and let `ExecutorMonitor`'s own removal handling take over), or
adjust the comment so the limitation is on the record.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && conf.get(DECOMMISSION_ENABLED)
Review Comment:
**Finding 6.** The stated requirement is "an external shuffle service to
keep the shuffle data of the decommissioned executors", but the check is
`spark.shuffle.service.enabled` specifically, which also rejects a deployment
that keeps shuffle data through a reliable-storage `ShuffleDataIO`.
`ExecutorAllocationManager.validateSettings` already treats the two as
alternatives (`!conf.get(SHUFFLE_SERVICE_ENABLED) && !reliableShuffleStorage`),
and `shuffleDriverComponents` is reachable from here:
```scala
private[spark] def executorHoldSupported: Boolean = {
schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
(conf.get(SHUFFLE_SERVICE_ENABLED) ||
_shuffleDriverComponents.supportsReliableStorage()) &&
conf.get(DECOMMISSION_ENABLED)
}
```
Being too strict fails silently here -- the control simply never renders and
`holdExecutors()` throws -- so it is easy to miss. For what it's worth, shuffle
tracking and `spark.storage.decommission.shuffleBlocks.enabled` are correctly
*not* alternatives: holding the whole application leaves no peer to migrate to.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
Review Comment:
**Finding 7.** Worth saying explicitly that cached blocks do not survive a
hold. Shuffle output does, via the shuffle service, but persisted RDD/DataFrame
blocks are gone: `BlockManagerDecommissioner` needs a live peer to replicate
to, and holding decommissions every executor at once, so
`rddBlockMigrationRunnable` takes its "No available peers to receive RDD
blocks, stop migration." branch (and with the default
`spark.storage.decommission.enabled=false` it never even tries). After resume,
everything cached is recomputed.
"pick up where it left off" here, and "no loss of in-progress task work or
shuffle output" in the PR description, are both accurate about tasks and
shuffle -- so one sentence in this scaladoc and in the `spark.ui.holdEnabled`
doc is all it needs, e.g. "Cached blocks are not preserved and are recomputed
after resuming."
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -175,6 +175,17 @@ 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
Review Comment:
**Finding 8.** Worth considering whether the hold can ride on the existing
target machinery instead of running beside it.
As written the feature adds a second control path: `suspended`
short-circuits `schedule()`, `suspend()`/`resume()` push to the cluster manager
themselves, `suspendedSyncPending` re-implements the retry `doUpdateRequest`
already has, `reset()` grows a special case and `onStageSubmitted` grows
another. Findings 1, 2 and 4 are all consequences of that path not inheriting
`doUpdateRequest`'s defensiveness.
The alternative is to keep `schedule()` running and make the *bounds* zero
while held:
```scala
private def effectiveMinNumExecutors = if (suspended) 0 else
minNumExecutors
private def effectiveMaxNumExecutors = if (suspended) 0 else
maxNumExecutors
// and maxNumExecutorsNeededPerResourceProfile returns 0 while suspended
```
`suspend()` then only sets the flag (plus `initializing = false`, so the
tick is not skipped) and talks to nobody. On the next tick
`updateAndSyncNumExecutorsTarget` sees `maxNeeded(0) < targetExecs`,
`decrementExecutors` drives each target to `math.max(0, 0) = 0`, and
`doUpdateRequest` does the push, checks the ack, rolls the target back on
failure and retries on the following tick. That is exactly what findings 1, 2
and 4 ask for, and it falls out for free:
- `reset()` restores `initialNumExecutors`, so the next tick decrements to 0
and pushes -- the `suspendedSyncPending` field and its `reset()`/`schedule()`
blocks go away, and the YARN AM-restart case is covered without them;
- a new resource profile in `onStageSubmitted` gets `initialNumExecutors`
and is decremented to 0 on the next tick -- that special case goes away too;
- both `if (!testing)` guards go away, so the production push is the one the
suite exercises;
- when the target is already 0 nothing is pushed at all, instead of an
unconditional request.
`resume()` keeps its floor restore and `addTime` nudge as-is.
Two honest counter-arguments. `schedule()` would also keep running
`timedOutExecutors()`/`removeExecutors()` while held, so idle executors get
drained by the normal idle path alongside your explicit `decommissionExecutors`
call, and "nothing happens while held" stops being visible as one `return`. And
`initializing` needs a decision: `suspend()` clearing it changes what
`resume()`'s `if (initializing) initialNumExecutors else minNumExecutors` floor
resolves to, so the "resume before the first stage" case needs the pre-hold
value remembered.
If you prefer the explicit path, that's fair -- then findings 1, 2 and 4
each need fixing on their own.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,126 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, an external shuffle
service to keep the
+ * shuffle data of the decommissioned executors, and decommission support.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ conf.get(SHUFFLE_SERVICE_ENABLED) && 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 through the
external shuffle
+ * service, so the application can later pick up where it left off via
`resumeExecutors()`.
+ *
+ * This requires both an external shuffle service
(`spark.shuffle.service.enabled`) and
+ * decommission support (`spark.decommission.enabled`).
+ *
+ * @return whether the request is received by the cluster manager.
Review Comment:
**Finding 5.** The supported branch returns a literal `true` regardless of
what the cluster manager said, so this `@return` does not hold -- same for
`resumeExecutors()`. The sibling `requestTotalExecutors` returns
`b.requestTotalExecutors(...)` directly, and the caller has no other way to
learn the request was rejected.
Once finding 2 makes the ack available, propagating it is cheap -- have
`suspend()`/`resume()` return the ack and use it:
```scala
val acknowledged = executorAllocationManager match {
case Some(manager) => manager.suspend()
case None => b.requestTotalExecutors(...)
}
...
acknowledged
```
Otherwise the doc should say the boolean only reports whether the current
scheduler supports holding.
##########
core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala:
##########
@@ -354,6 +354,37 @@ 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)
+ if (parent.sc.get.executorsHeld) {
+ val numDraining = parent.sc.get.getExecutorIds().size
+ val status = if (numDraining > 0) {
+ s"Held (draining $numDraining executors)"
Review Comment:
**Finding 10.** `Held (draining 1 executors)` -- visible in the state
diagram in the PR description too.
```suggestion
s"Held (draining $numDraining executor${if (numDraining >
1) "s" else ""})"
```
--
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]
