dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3805781100
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -175,6 +175,17 @@ private[spark] class ExecutorAllocationManager(
// (2) an executor idle timeout has elapsed.
@volatile private var initializing: Boolean = true
+ // Whether allocation is suspended because the executors are held. While
this is true,
+ // `schedule()` is a no-op so that pending tasks do not bring up new
executors.
+ // See `SparkContext.holdExecutors()`.
+ private var suspended: Boolean = false
Review Comment:
Thank you for the thorough analysis. I kept the explicit path for now: the
static-allocation side needs findings 1/2/4 handled individually anyway (there
is no schedule loop to inherit the defensiveness of `doUpdateRequest`), and
capping the bounds would change what the EAM metrics report while held, keep
the idle reaper running alongside the explicit drain, and require remembering
the pre-hold `initializing` value as you noted. With 1, 2 and 4 fixed directly
in c577f4d (shared `syncTargetsWithClient()` + `targetSyncPending` retry, and
the held re-assert in `CoarseGrainedSchedulerBackend.reset()`), the explicit
path now has the same defensiveness. I am happy to revisit the bounds-based
design as a follow-up refactor.
##########
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) {
+ // The executors are held; drain this late-registered executor
immediately.
+ val decommissioned = decommissionExecutors(
+ Array((executorId, ExecutorDecommissionInfo("Executors are
held"))),
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ scheduler.sc.executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
Review Comment:
Right -- dropped the call in c577f4d and left a comment on the record that
decommissioning of late registrants is under-reported in the metrics; the
monitor own removal handling covers the eventual exit.
##########
core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala:
##########
@@ -354,6 +354,37 @@ private[ui] class AllJobsPage(parent: JobsTab, store:
AppStatusStore) extends We
<strong>Scheduling Mode: </strong>
{schedulingMode}
</li>
+ {
+ if (parent.holdEnabled &&
parent.sc.exists(_.executorHoldSupported)) {
+ val basePathUri = UIUtils.prependBaseUri(request,
parent.basePath)
+ if (parent.sc.get.executorsHeld) {
+ val numDraining = parent.sc.get.getExecutorIds().size
+ val status = if (numDraining > 0) {
+ s"Held (draining $numDraining executors)"
Review Comment:
Fixed in c577f4d.
--
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]
