dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3815686116
##########
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:
Fixed in a927b1e with the deferral you sketched: `handleJobSubmitted` now
checks the (seamed) hold state before the terminal gang admission and re-posts
the submission on the `timeIntervalNumTasksCheck` timer while held, so a
pipelined job submitted during a hold waits for the resume like a barrier job.
Added a `DAGSchedulerSuite` test through the new `executorsHeld` seam asserting
the job is neither failed nor staged while held.
##########
core/src/test/scala/org/apache/spark/SparkContextSuite.scala:
##########
@@ -1578,6 +1584,97 @@ 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))
+ }
+
+ private def verifyHoldAndResumeExecutors(conf: SparkConf): Unit = {
+ // The executors register with the external shuffle service on startup, so
run one
+ val transportConf = SparkTransportConf.fromSparkConf(conf, "shuffle",
numUsableCores = 2)
+ val rpcHandler = new ExternalBlockHandler(transportConf, null)
+ val transportContext = new TransportContext(transportConf, rpcHandler)
+ val server = transportContext.createServer()
+ try {
+ conf.set(SHUFFLE_SERVICE_PORT, server.getPort)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 1, 60000)
+ assert(sc.executorHoldSupported)
+ assert(!sc.executorsHeld)
+
+ assert(sc.holdExecutors())
Review Comment:
Added in a927b1e. The e2e helper now pins both guarantees with your oracle:
it runs a 4-map-task shuffle before the hold, asserts the 4 map outputs survive
the drain, and after resume re-reads the shuffle under a stage-submitted
listener asserting only the reduce stage runs (checked after the job, given the
asynchronous executor-removal processing you pointed out). For the
running-tasks half there is a new test in the `WorkerDecommissionSuite` shape:
two tasks on one core, hold while the first is running -- the executor drains
only after that task finishes, and the still-pending second task runs after
resume, completing the job with no lost work.
--
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]
