dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3805776870
##########
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:
Fixed in c577f4d. `suspend()` now catches internally and returns the ack,
and the static path wraps `requestTotalExecutors` in a `NonFatal` catch, so the
drain below runs unconditionally.
##########
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:
Fixed in c577f4d. Generalized the flag to `targetSyncPending` with a shared
`syncTargetsWithClient()`: `suspend()`, `resume()` and `reset()` all arm it
when a push fails or is rejected, and `schedule()` retries it on every tick,
suspended or not, until acknowledged -- which also covers the `resume()` mirror
hole you described. Added a test that a rejected suspend push is re-pushed on
the next tick and stops once acknowledged.
##########
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:
Added in c577f4d, modeled on the SPARK-41766 test as suggested.
--
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]
