dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3806532976
##########
core/src/test/scala/org/apache/spark/SparkContextSuite.scala:
##########
@@ -1578,6 +1581,60 @@ class SparkContextSuite extends SparkFunSuite with
LocalSparkContext with Eventu
assert(err.getMessage.contains("Int.MaxValue"))
assert(err.getMessage.contains("overflowed"))
}
+
+ test("SPARK-58828: holdExecutors and resumeExecutors are unsupported by the
local scheduler") {
+ sc = new SparkContext(new
SparkConf().setAppName("test").setMaster("local"))
+ assert(!sc.executorHoldSupported)
+ assert(!sc.holdExecutors())
+ assert(!sc.resumeExecutors())
+ }
+
+ test("SPARK-58828: holdExecutors requires external shuffle service and
decommission support") {
+ sc = new SparkContext(
+ new SparkConf().setAppName("test").setMaster("local-cluster[1,1,1024]"))
+ assert(!sc.executorHoldSupported)
+ val err = intercept[IllegalArgumentException] {
+ sc.holdExecutors()
+ }
+ assert(err.getMessage.contains(SHUFFLE_SERVICE_ENABLED.key))
+ assert(err.getMessage.contains(DECOMMISSION_ENABLED.key))
+ }
+
+ test("SPARK-58828: holdExecutors drains the executors and resumeExecutors
brings them back") {
Review Comment:
Added in 137ebdb: extracted the setup into
`verifyHoldAndResumeExecutors(conf)` and added a second case with
`DYN_ALLOCATION_ENABLED=true, INITIAL/MIN=1`, so `manager.suspend()` and the
`executorsDecommissioned` call now run end to end. One detail: since this test
never submits a stage, `initializing` stays true and the resume floor is
actually `initialNumExecutors` rather than `minNumExecutors` -- with both set
to 1 as you suggested, the assertions hold either way.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,159 @@ 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 through the
external shuffle
+ * service, 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. 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,
Review Comment:
Fixed in 137ebdb as suggested: added a `private[spark]
requestedTotalExecutors()` snapshot next to `setExecutorsHeld`, and
`heldNumExecutors` now prefers it when non-empty, keeping the
standalone/conf/default heuristic only for the never-pushed case.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -339,6 +435,16 @@ private[spark] class ExecutorAllocationManager(
* This is factored out into its own method for testing.
*/
private def schedule(): Unit = synchronized {
+ if (targetSyncPending) {
Review Comment:
Fixed in 137ebdb with the tick-counter scheme you sketched: the first
attempt after arming stays immediate (so both retry tests hold as written),
each failure doubles the delay up to a cap of 100 ticks (10s), and success
resets it. Added a test asserting the second retry waits a tick.
--
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]
