sunchao opened a new pull request, #58131: URL: https://github.com/apache/spark/pull/58131
### Why are the changes needed? [SPARK-58879](https://issues.apache.org/jira/browse/SPARK-58879) is the graceful-decommission equivalent of [SPARK-9552](https://issues.apache.org/jira/browse/SPARK-9552). Dynamic allocation decides which executors have timed out using asynchronously updated idle information. An executor can receive new work between that decision and the request to remove it. The ordinary kill path handles this by checking whether the executor is still idle when `force = false`; the graceful path currently checks only whether the executor is active. For example, dynamic allocation can select executor E for idle removal just as a new stage arrives. The scheduler assigns a task to E and records it as running, but has not yet delivered `LaunchTask`. The graceful scale-down request then accepts E and starts decommissioning it. The already-assigned task can arrive after decommissioning has begun. This needlessly drains a busy executor and can cause task retries, including failed block writes when storage decommissioning is enabled. The missing guarantee is narrow: an ordinary idle-removal request must not accept an executor that has acquired new task assignments. Explicit host drains and other forced decommission requests still need to be able to retire busy executors. ### What changes were proposed in this PR? Core dynamic allocation now uses a separate idle-only decommission entry point. The coarse-grained backend checks the scheduler's current task assignments while holding the same scheduler/backend locks used to make offers, and keeps those locks until the existing decommission path has made the accepted executors unavailable for new offers. That gives the race two safe outcomes. If task assignment happens first, the removal request skips the busy executor and can try again later. If decommissioning happens first, subsequent offers cannot select that executor. The check uses scheduler bookkeeping, which already includes assigned tasks whose `LaunchTask` message has not been sent yet. The existing unconditional decommission method is unchanged. The new entry point delegates to it with only active, idle, deduplicated executor IDs, preserving cluster-manager overrides such as Kubernetes pod metadata updates. Clients that cannot provide atomic idle admission decline the new operation rather than invoking unconditional decommissioning. This adds no configuration or executor RPC protocol change. ### How was this PR tested? Added deterministic regressions using a real `TaskSchedulerImpl` for both race orderings, ordinary and barrier tasks, rejected/duplicate requests, and explicit busy-executor decommissioning. Allocation-manager tests check accepted-subset accounting and later retries; Kubernetes tests check that only accepted IDs reach the existing metadata path. On JDK 17, all 83 tests in the three focused suites passed (68 core and 15 Kubernetes), along with the four compile/test style checks: ```sh build/sbt -Djava.net.preferIPv4Stack=true -Pkubernetes \ 'set LocalProject("core") / Test / javaOptions += "-Djava.net.preferIPv4Stack=true"' \ 'set LocalProject("kubernetes") / Test / javaOptions += "-Djava.net.preferIPv4Stack=true"' \ 'core/testOnly org.apache.spark.scheduler.CoarseGrainedSchedulerBackendSuite org.apache.spark.ExecutorAllocationManagerSuite' \ 'kubernetes/testOnly org.apache.spark.scheduler.cluster.k8s.KubernetesClusterSchedulerBackendSuite' \ 'core/scalaStyleOnCompile' 'core/scalaStyleOnTest' \ 'kubernetes/scalaStyleOnCompile' 'kubernetes/scalaStyleOnTest' ``` As a negative control, removing only `!scheduler.isExecutorBusy(executorId)` made both assignment-before-`LaunchTask` tests fail because the busy executor was accepted. The predicate was restored, its source checksum verified, and all 83 focused tests and four style checks passed again. The IPv4 option is a local test-runner workaround, not a Spark configuration change in this PR. Full Kubernetes integration tests were not run because no local Kubernetes test cluster was available. ### Does this PR introduce _any_ user-facing change? Yes. With core dynamic allocation and graceful decommissioning enabled, idle scale-down no longer retires an executor merely because an earlier idle observation became stale after new task assignment. Explicit and infrastructure-initiated decommissioning retain their existing behavior. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex desktop (version 26.814.41957) -- 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]
