dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3808324568
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -321,6 +326,27 @@ 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. Note
+ // that ExecutorMonitor cannot be told here: it builds its state
asynchronously
+ // from the SparkListenerExecutorAdded event posted above, so this
executor is not
+ // tracked there yet and its decommissioning is under-reported in
the metrics.
+ // The monitor's own executor-removal handling covers the eventual
exit.
+ decommissionExecutors(
+ Array((executorId, ExecutorDecommissionInfo("Executors are
held"))),
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ // The cluster manager granted this executor against a stale
requirement (e.g. a
+ // restarted AM using its own initial target, or a lost response
to an earlier
+ // request), so re-assert the zero requirement, or it would keep
granting
+ // replacements that churn through this drain. Do not await the
response: this
+ // runs inside the RPC handler.
+ CoarseGrainedSchedulerBackend.this.synchronized {
+ val defaultProf =
scheduler.sc.resourceProfileManager.defaultResourceProfile
+ requestedTotalExecutorsPerResourceProfile(defaultProf) = 0
+
doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
Review Comment:
Fixed in 7dee2f0. The reassertion is extracted into
`reassertHeldRequirement()`, which re-checks `executorsHeld` under the backend
lock before touching the requirement. Since `resumeExecutors()` clears the flag
before restoring the requirement, every interleaving now converges on the
restored value: a reassertion that runs before the clear is overwritten by the
resume push, and one that runs after sees the flag down and skips. Deliberately
not covered: the registrant-drain check itself can still race a concurrent
resume, which costs one executor churn that the restored requirement
immediately replaces (folding the decommission under the same lock would invert
the `withLock` scheduler-then-backend order). Added a test that a stale
reassertion after the hold is lifted leaves the requirement alone while a held
one zeroes it.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ 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
Review Comment:
Good catch, thank you -- fixed in 7dee2f0. Both fields moved to the early
private-variables section so their initializers run before the constructor
attaches the UI handlers and enters `postStartHook()`, with a comment
explaining why they must stay there.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ 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 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.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+ // Floor with the live executor count: killing an executor of
a static
+ // application seeds the tracked requirement with a zero,
which is a
+ // bookkeeping artifact rather than a requirement to restore.
+ math.max(
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0),
+ b.getExecutorIds().size)
Review Comment:
Fixed in 7dee2f0 as suggested. The backend now tracks whether a total was
ever explicitly requested (`requestExecutors` / `requestTotalExecutors` set it;
the `adjustExecutors` bookkeeping seed does not), and `holdExecutors()` uses
the tracked requirement only in that case -- so an explicit 2 (or 0) is
restored faithfully, the `math.max` live-count floor is gone, and the
kill-seeded zero still falls back to the standalone/conf/default heuristic.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ 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 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.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+ // Floor with the live executor count: killing an executor of
a static
+ // application seeds the tracked requirement with a zero,
which is a
+ // bookkeeping artifact rather than a requirement to restore.
+ math.max(
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0),
+ b.getExecutorIds().size)
+ } 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,
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
+ }
+ }
+
+ // Restore the hold invariant without dynamic allocation: push the zero
executor requirement
+ // and drain the current executors. The requirement push must not abort the
drain, which has
+ // to run even when the cluster manager is temporarily unreachable; the
registration guard
+ // in CoarseGrainedSchedulerBackend remains the backstop when the push fails.
+ private[spark] def zeroExecutorRequirementAndDrain(b:
ExecutorAllocationClient): Boolean = {
+ val acknowledged = try {
+ b.requestTotalExecutors(
+ immutable.Map(resourceProfileManager.defaultResourceProfile.id -> 0),
+ immutable.Map.empty, immutable.Map.empty)
Review Comment:
Fixed in 7dee2f0 with a capability check at the backend layer, as suggested:
`CoarseGrainedSchedulerBackend.supportsExecutorHold` defaults to true and
`KubernetesClusterSchedulerBackend` overrides it to allow holding only with the
`direct` allocator (custom allocator classes are conservatively excluded too,
since their scale-down semantics are unknown). `executorHoldSupported` includes
the check, so on such deployments `holdExecutors()` is rejected and the UI
control does not render. Controller-aware draining for the
Deployment/StatefulSet allocators is left as a follow-up.
--
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]
