peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3806029881
##########
core/src/test/scala/org/apache/spark/SparkContextSuite.scala:
##########
@@ -1578,6 +1581,60 @@ 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))
+ }
+
+ test("SPARK-58828: holdExecutors drains the executors and resumeExecutors
brings them back") {
Review Comment:
**Finding 13.** This is the only end-to-end test of the feature, and it runs
the static-allocation path (`executorAllocationManager.isEmpty`). Nothing
exercises `holdExecutors()`/`resumeExecutors()` with dynamic allocation on, so
the branch the PR description leads with has never run: `case Some(manager) =>
manager.suspend()` at
`core/src/main/scala/org/apache/spark/SparkContext.scala:2148`, and the
`executorMonitor.executorsDecommissioned(decommissioned)` call at `:2169`.
`ExecutorAllocationManagerSuite` covers `suspend()`/`resume()` directly against
a mocked client, which is a different thing — it cannot catch the glue, and the
glue is where findings 1 and 5 lived.
The delta looks small. `WorkerDecommissionExtendedSuite` already runs
dynamic allocation on `local-cluster` with `spark.decommission.enabled`, so the
shape works, and this test already stands up the shuffle service. A second case
reusing the same setup:
```scala
conf.set(SHUFFLE_SERVICE_PORT, server.getPort)
.set(DYN_ALLOCATION_ENABLED, true)
.set(DYN_ALLOCATION_INITIAL_EXECUTORS, 1)
.set(DYN_ALLOCATION_MIN_EXECUTORS, 1)
```
with the same assertions as here. `DYN_ALLOCATION_MIN_EXECUTORS = 1` is the
part that matters: `resume()`'s floor is `minNumExecutors` once a stage has
been submitted, so with the default `0` an idle application comes back with a
zero target and the `getExecutorIds().nonEmpty` assertion would need a job to
drive the ramp-up instead.
##########
docs/configuration.md:
##########
@@ -1605,6 +1605,19 @@ 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>
Review Comment:
**Finding 12.** The config is documented here, but the thing users actually
see — the `(hold)` / `(resume)` control on the Jobs page — is not in
`docs/web-ui.md`, and that file already documents the sibling control:
> `docs/web-ui.md:102` — "An active stage shows a small **(kill)** link next
to its description; clicking it asks Spark to cancel that stage."
The `## Jobs Tab` section is where it belongs. A couple of sentences on the
same model would do it, e.g.
> 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. The control only appears
when
> `spark.ui.holdEnabled` is true, `spark.decommission.enabled` is true and
shuffle data is kept
> outside the executors — see [Configuration](configuration.html#spark-ui).
The prerequisites are worth naming there in particular: when they aren't met
the control simply doesn't render, with nothing to tell the user why.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -339,6 +435,16 @@ private[spark] class ExecutorAllocationManager(
* This is factored out into its own method for testing.
*/
private def schedule(): Unit = synchronized {
+ if (targetSyncPending) {
Review Comment:
**Finding 15.** The retry fires on every 100 ms tick with nothing in
between, and the failure it exists for lasts minutes.
The case named in the comment — "before the YARN AM registered" — is
rejected immediately and cheaply: `YarnSchedulerBackend`'s endpoint does
`logWarning("Attempted to request executors before the AM has registered!");
context.reply(false)`
(`resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala:373-375`).
So for as long as the AM is down this is 10 rejected asks and 10 warning lines
per second, for a condition that resolves on AM-restart timescales.
The other mode is the one that concerns me more. When the ask times out
rather than replying, `client.requestTotalExecutors` ends in
`defaultAskTimeout.awaitResult(...)` inside `CoarseGrainedSchedulerBackend`,
which blocks for up to `spark.network.timeout` (120 s by default) — and
`schedule()` holds the `allocationManager` monitor throughout. That is the same
monitor every `ExecutorAllocationListener` callback takes (`onStageSubmitted`,
`onTaskStart`, `onTaskEnd`, ...), so the management listener queue stalls for
the duration. `doUpdateRequest` has the same exposure, but only on a tick where
a target actually changed, whereas this fires unconditionally while pending.
A tick counter with a cap is enough — no need for a real scheduler:
```scala
if (targetSyncPending && ticksUntilTargetSync <= 0) {
// Back off: the conditions this retries (AM restart, unreachable
cluster manager) resolve
// on a timescale of minutes, not of the 100ms allocation tick.
syncTargetsWithClient()
} else if (targetSyncPending) {
ticksUntilTargetSync -= 1
}
```
with `syncTargetsWithClient()` setting `ticksUntilTargetSync = 0` on success
and doubling it up to a cap on failure. Arming it at `0` keeps the first tick
after `targetSyncPending` is set an immediate push, so both new retry tests
still hold as written.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -175,6 +175,18 @@ 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.
+ private var targetSyncPending: Boolean = false
Review Comment:
**Finding 18.** The class doc still states the opposite of this field —
`core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:71-72`:
> 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.
`targetSyncPending` is exactly retry logic, and it exists because that
assumption does not hold for the zero target a hold depends on. Worth a clause
up there so the next reader doesn't trust the old sentence.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,159 @@ 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, 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.isInstanceOf[ExecutorAllocationClient] &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
Review Comment:
**Finding 11.** The description hasn't caught up with two of the `c577f4d`
fixes, and both statements are now wrong in a way a reader would act on.
> This requires both `spark.shuffle.service.enabled` (to keep the shuffle
data of the decommissioned executors) and `spark.decommission.enabled`, and is
rejected otherwise.
After the finding-6 fix this line *is* the check, and it is a disjunction: a
`ShuffleDataIO` with reliable storage is an accepted alternative, so an
ESS-less deployment is not rejected. The scaladoc at
`core/src/main/scala/org/apache/spark/SparkContext.scala:2106`, the `require`
message and the `spark.ui.holdEnabled` doc all state it correctly; only the
description still says "both".
> Executors are drained gracefully with no loss of in-progress task work or
shuffle output, and the application can be resumed later without recomputation.
The second clause is contradicted by the sentence you added at
`core/src/main/scala/org/apache/spark/SparkContext.scala:2104` — "Cached blocks
are not preserved and are recomputed after resuming." Shuffle output survives;
cached data is recomputed. Worth dropping "without recomputation" or qualifying
it to shuffle output, since it's the sentence someone will quote when deciding
whether a hold is safe for their workload.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,159 @@ 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, 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.isInstanceOf[ExecutorAllocationClient] &&
+ (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 through the
external shuffle
Review Comment:
**Finding 17.** This sentence still names the external shuffle service as
the only mechanism, two lines above the paragraph you rewrote for finding 6 to
say it can also be a `ShuffleDataIO` with reliable storage. The
`spark.ui.holdEnabled` doc and `configuration.md` already avoid this.
```scala
* 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()`.
```
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,159 @@ 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, 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.isInstanceOf[ExecutorAllocationClient] &&
+ (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 through the
external shuffle
+ * service, 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.
+ *
+ * @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 b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ synchronized {
+ if (_executorsHeld) {
+ true
+ } else {
+ 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,
Review Comment:
**Finding 14.** `spark.executor.instances` wins over the requirement the
application actually has, so a hold can permanently shrink an application that
was scaled up through the `requestExecutors` API — silently, with nothing in
the log to say so.
Concretely, with dynamic allocation off, `spark.executor.instances = 10` and
a `sc.requestExecutors(40)`:
`CoarseGrainedSchedulerBackend.requestedTotalExecutorsPerResourceProfile` holds
50 and 50 executors are running. `holdExecutors()` takes
`conf.get(EXECUTOR_INSTANCES)`, so `heldNumExecutors = 10`, and
`resumeExecutors()` pushes 10. The application comes back at a fifth of its
size. The fallback below it (`math.max(getExecutorIds().size,
DEFAULT_NUMBER_EXECUTORS)`) would have got this right — it's only skipped
because the conf happens to be set.
The backend already knows the exact answer, so it can be read instead of
reconstructed, keeping this heuristic for the case where the map is empty (a
standalone application that never pushed a requirement, which is what the
`Int.MaxValue` above is for):
```scala
heldNumExecutors = b match {
case cg: CoarseGrainedSchedulerBackend
if cg.requestedTotalExecutors().nonEmpty =>
cg.requestedTotalExecutors()
.getOrElse(resourceProfileManager.defaultResourceProfile, 0)
case _: StandaloneSchedulerBackend => Int.MaxValue
case _ =>
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.getExecutorIds().size,
SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS))
}
```
`requestedTotalExecutorsPerResourceProfile` is `private`
(`core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:88`),
so this needs a small `private[spark]` snapshot accessor next to
`setExecutorsHeld`.
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +64,32 @@ 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.
+ private lazy val holdRequestExecutor =
+ ThreadUtils.newDaemonSingleThreadExecutor("spark-ui-hold-resume")
Review Comment:
**Finding 16.** Nothing ever shuts this executor down, so the
`spark-ui-hold-resume` thread outlives the `SparkContext`.
`ThreadUtils.newDaemonSingleThreadExecutor` starts its thread on the first
`execute`, and there is no `shutdown` anywhere: `WebUI.stop()` only stops the
Jetty server (`core/src/main/scala/org/apache/spark/ui/WebUI.scala:181-185`)
and `SparkUITab` has no lifecycle hook. In a JVM that creates and stops several
`SparkContext`s — which is what every core test suite does — each one that
served a hold or resume request leaks a thread. They are daemons so nothing
fails today, but `SparkFunSuite`'s thread audit would flag it the moment a test
drives the control, which finding 13 makes more likely.
`SparkUI` already overrides `stop()`, so the hook is nearly there —
`jobsTab` is currently a local `val` in `initialize()`
(`core/src/main/scala/org/apache/spark/ui/SparkUI.scala:101`) and would need
hoisting to a field:
```scala
override def stop(): Unit = {
super.stop()
jobsTab.stop()
logInfo(log"Stopped Spark web UI at ${MDC(WEB_URL, webUrl)}")
}
```
with `JobsTab.stop()` shutting the executor down only if it was ever
created, so the `lazy val` isn't forced on every UI teardown. Alternatively
drop the dedicated executor and submit onto an existing pool.
--
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]
