dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3816277146
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +66,87 @@ 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. AllJobsPage renders it only next to the matching control, so a
stale message
+ // never shows against the opposite operation.
+ @volatile private var holdRequestStatus: Option[(Boolean, String)] = None
+
+ private[jobs] def lastHoldRequestStatus(isHold: Boolean): Option[String] =
+ holdRequestStatus.collect { case (`isHold`, message) => message }
Review Comment:
Fixed in 1614451 as suggested: the status renders whenever it is set,
labelled with the operation it belongs to (`(hold: ...)` / `(resume: ...)`),
and the `isHold` gating argument is gone from `AllJobsPage` -- so a rejected
publish is reported even though the page already shows the `(resume)` control.
##########
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:
Your correction to my rationale is right, for the record: the publish has
nothing to do with the abort window -- what makes a monitor-scoped check
unattractive is that a pipelined group discovered at that point would require
rolling back the already-set held flags and the published zeros. Same
conclusion, and the check stays belt-and-braces behind the submit-side deferral.
--
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]
