dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3825243709
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,232 @@ 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 -- unless `spark.executor.decommission.forceKillTimeout` is set, in
which case an
+ * executor still running tasks is killed after that timeout. 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. While held the application has no executors,
so with
+ * `spark.default.parallelism` unset the default parallelism falls back to
2, and an RDD
+ * created during the hold keeps that partition count after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`),
shuffle data kept
+ * outside the executors -- either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage -- and a
+ * scheduler backend that can hold executors: Standalone, YARN, and
Kubernetes with the
+ * `direct` pods allocator. Fallback storage
+ * (`spark.storage.decommission.fallbackStorage.path`) deliberately does not
qualify:
+ * shuffle blocks not yet migrated when an executor exits are dropped.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * Pipelined-shuffle jobs are outside the hold's scope. A 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, and a group that
slips past the
+ * check is aborted rather than drained, since a pipelined task set
tolerates no task
+ * failure. A pipelined job submitted while held fails its gang admission
immediately
Review Comment:
Fixed in 44baf1d with your suggested clause, in all three places (the
scaladoc, `web-ui.md`, and the PR description): with the internal
`spark.scheduler.pipelinedGroup.slotCheck.enabled=false` there is no admission
check, so a pipelined job submitted while held waits for the resume instead of
failing.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2093,8 +2093,15 @@ private[spark] class DAGScheduler(
} catch {
case e: BarrierJobSlotsNumberCheckFailed =>
// If jobId doesn't exist in the map, Scala coverts its value null to
0: Int automatically.
- val numCheckFailures =
barrierJobIdToNumTasksCheckFailures.compute(jobId,
- (_: Int, value: Int) => value + 1)
+ // Do not consume the retry budget while the executors are held: the
slot check sees
+ // zero slots for the whole hold, and the job should wait for the
resume like any
+ // other job instead of failing when the retries run out.
+ val numCheckFailures = if (sc.executorsHeld) {
Review Comment:
A scope correction to my earlier replies on this thread: the deferral
described above was removed in 10732ea. Keeping a budget-less pipelined re-post
correct across the resume boundary kept accreting edge cases (your findings
29/30/34/35), so the PR now takes the documented-exclusion alternative you
offered when you opened this finding: pipelined-shuffle jobs are explicitly
outside the hold's scope. A hold is still rejected while a pipelined job is
running (`hasPipelinedTaskSets`, now landed separately as SPARK-58913), and a
pipelined job submitted while held fails its gang admission immediately --
before any stage is created, so no partial scheduler state -- with the
scaladoc, `web-ui.md` and the description saying to resubmit it after the
resume. The barrier retry-budget freeze is unchanged. @viirya re-reviewed the
simplification at 10732ea and agreed with the direction.
--
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]
