dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3813870052
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +66,44 @@ 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 def holdRequestExecutorPool: ExecutorService = synchronized {
+ holdRequestExecutor.getOrElse {
+ val pool =
ThreadUtils.newDaemonSingleThreadExecutor("spark-ui-hold-resume")
+ holdRequestExecutor = Some(pool)
+ pool
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ holdRequestExecutor.foreach(_.shutdownNow())
+ }
+
+ def handleHoldRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.execute(() =>
Utils.tryLogNonFatalError(ctx.holdExecutors()))
+ // Do a quick pause here to give Spark time to decommission the
executors so the
+ // application shows up as held after the refresh. Note that this will
block the
+ // serving thread so the time should be limited in duration.
+ Thread.sleep(100)
+ }
+ }
+ }
+
+ def handleResumeRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.execute(() =>
Utils.tryLogNonFatalError(ctx.resumeExecutors()))
Review Comment:
Fixed in 8bfca77 along the lines you suggested: `JobsTab` keeps a
`@volatile` status for the last hold/resume request (`hold requested` / `resume
requested` while running, `not acknowledged, see the driver logs` on failure,
cleared on success), `AllJobsPage` renders it next to the Application line, and
both `Thread.sleep(100)` calls are gone.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,201 @@ 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 _: ExecutorAllocationClient => true
+ 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.
+ *
+ * @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: prefer what the
application actually
+ // requested (e.g. through requestExecutors), which the backend
tracks. When it
+ // never pushed a requirement, Standalone has no explicit one 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.
+ // Use the tracked requirement only when it was explicitly
requested: killing an
+ // executor of a static application seeds it with a bookkeeping
zero, which is
+ // not a requirement to restore.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend if
cg.hasExplicitExecutorRequests =>
+ cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0)
+ } else {
+ 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) =>
+ val acknowledged = manager.suspend()
+ drainHeldExecutors(b)
+ acknowledged
+ case None => zeroExecutorRequirementAndDrain(b)
+ }
+ }
+ }
+ 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,
Review Comment:
Fixed in 8bfca77 with the first option: while the executors are held,
`DAGScheduler` no longer consumes the barrier retry budget (the failure count
is read but not incremented), so a barrier job keeps re-checking every 15s and
proceeds after resume like any other job. `BarrierStageOnSubmittedSuite` passes
for regressions.
--
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]
