You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by su...@apache.org on 2026/08/20 15:59:03 UTC
(spark) branch branch-4.x updated: [SPARK-58879][CORE] Avoid decommissioning executors that became busy
This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 083df6652a7f [SPARK-58879][CORE] Avoid decommissioning executors that became busy
083df6652a7f is described below
commit 083df6652a7fa7ac046eb6bb2d373775c48bcec6
Author: Chao Sun <ch...@openai.com>
AuthorDate: Thu Aug 20 08:58:40 2026 -0700
[SPARK-58879][CORE] Avoid decommissioning executors that became busy
### 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 curr [...]
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. Idle-only requests reject unknown or already-retiring executors without queuing a deferred decommission; explicit requests retain their existing [pre-registration replay behavior](https://issues.apache.org/jira/browse/SPARK-41766).
The new method's default declines all requests rather than invoking unconditional decommissioning. All in-tree clients inherit the coarse-grained implementation; custom clients that implement the allocation trait directly must override the new method to support graceful dynamic-allocation scale-down. This preserves the old three-argument method and its overrides without silently substituting abrupt removal for a custom graceful implementation. This adds no configuration or executor RP [...]
### 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. A second, test-only negative control added an executor-ID selector containing both the accepted and rejected IDs; the focused Kubernetes assertion failed as expected. Both temporary changes were restored, their source checksums verified, and all 83 focused tests and four style checks passed again. The IPv4 option is a [...]
### 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: Codex
Closes #58131 from sunchao/dev/chao/codex/spark-58879-idle-decommission.
Authored-by: Chao Sun <ch...@openai.com>
Signed-off-by: Chao Sun <ch...@openai.com>
(cherry picked from commit b1ce1292b62601a12a5203bbcd9e09fad49a6cfc)
Signed-off-by: Chao Sun <ch...@openai.com>
---
.../apache/spark/ExecutorAllocationClient.scala | 13 +
.../apache/spark/ExecutorAllocationManager.scala | 5 +-
.../cluster/CoarseGrainedSchedulerBackend.scala | 15 ++
.../spark/ExecutorAllocationManagerSuite.scala | 50 +++-
.../CoarseGrainedSchedulerBackendSuite.scala | 290 ++++++++++++++++++++-
.../KubernetesClusterSchedulerBackendSuite.scala | 143 +++++++++-
6 files changed, 496 insertions(+), 20 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala
index c533f5cbc0b0..870a24a9b3b4 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala
@@ -104,6 +104,19 @@ private[spark] trait ExecutorAllocationClient {
countFailures = false)
}
+ /**
+ * Decommission only executors which are still idle when the request is accepted.
+ * Implementations must check for assigned tasks and stop further task assignment atomically.
+ * The default declines all requests. Clients that support graceful dynamic-allocation
+ * scale-down must override this method and provide the atomicity guarantee.
+ *
+ * @param executorsAndDecomInfo identifiers of executors and decommission information
+ * @param adjustTargetNumExecutors whether to reduce the target number of executors
+ * @return the ids of the executors accepted for decommissioning
+ */
+ def decommissionExecutorsIfIdle(
+ executorsAndDecomInfo: Array[(String, ExecutorDecommissionInfo)],
+ adjustTargetNumExecutors: Boolean): Seq[String] = Seq.empty
/**
* Request that the cluster manager decommission the specified executor.
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index 57f2d21bc7b7..559b3fcf3712 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -584,10 +584,9 @@ private[spark] class ExecutorAllocationManager(
if (decommissionEnabled) {
val executorIdsWithoutHostLoss = executorIdsToBeRemoved.map(
id => (id, ExecutorDecommissionInfo("spark scale down"))).toArray
- client.decommissionExecutors(
+ client.decommissionExecutorsIfIdle(
executorIdsWithoutHostLoss,
- adjustTargetNumExecutors = false,
- triggeredByExecutor = false)
+ adjustTargetNumExecutors = false)
} else {
client.killExecutors(executorIdsToBeRemoved.toSeq, adjustTargetNumExecutors = false,
countFailures = false, force = false)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
index e07cde82ffa0..2f957aead71d 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -624,6 +624,21 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
executorsToDecommission.toImmutableArraySeq
}
+ override def decommissionExecutorsIfIdle(
+ executorsAndDecomInfo: Array[(String, ExecutorDecommissionInfo)],
+ adjustTargetNumExecutors: Boolean): Seq[String] = withLock {
+ val idleExecutors = executorsAndDecomInfo.distinctBy(_._1).filter { case (executorId, _) =>
+ isExecutorActive(executorId) && !scheduler.isExecutorBusy(executorId)
+ }
+ if (idleExecutors.isEmpty) {
+ Seq.empty
+ } else {
+ // Keep both locks until the existing path marks these executors pending decommission.
+ // Use virtual dispatch so cluster-manager overrides receive only the filtered IDs.
+ decommissionExecutors(idleExecutors, adjustTargetNumExecutors, triggeredByExecutor = false)
+ }
+ }
+
override def start(): Unit = {
setupTokenManager()
setupUserCredentialManager()
diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
index 0872d39df535..7f622dc91772 100644
--- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala
@@ -21,7 +21,8 @@ import java.util.concurrent.TimeUnit
import scala.collection.mutable
-import org.mockito.ArgumentMatchers.any
+import org.mockito.ArgumentCaptor
+import org.mockito.ArgumentMatchers.{any, anyBoolean, eq => mockitoEq}
import org.mockito.Mockito._
import org.scalatest.PrivateMethodTester
@@ -1070,6 +1071,53 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite {
assert(manager.executorAllocationManagerSource.exitedUnexpectedly.getCount() === 1)
}
+ test("SPARK-58879: dynamic allocation accounts only for accepted idle decommissions") {
+ val conf = createConf(0, 5, 0, decommissioningEnabled = true)
+ .set(config.DYN_ALLOCATION_TESTING, false)
+ when(client.requestTotalExecutors(any(), any(), any())).thenReturn(true)
+ when(client.decommissionExecutorsIfIdle(any(), anyBoolean()))
+ .thenReturn(Seq("executor-2"))
+ val manager = createManager(conf)
+ onExecutorAddedDefaultProfile(manager, "executor-1")
+ onExecutorAddedDefaultProfile(manager, "executor-2")
+
+ assert(removeExecutorsDefaultProfile(manager, Seq("executor-1", "executor-2")) ===
+ Seq("executor-2"))
+ assert(executorsDecommissioning(manager) === Set("executor-2"))
+ assert(executorsPendingToRemove(manager).isEmpty)
+
+ val requests = ArgumentCaptor.forClass(
+ classOf[Array[(String, ExecutorDecommissionInfo)]])
+ verify(client).decommissionExecutorsIfIdle(requests.capture(), mockitoEq(false))
+ assert(requests.getValue.toSeq === Seq(
+ "executor-1" -> ExecutorDecommissionInfo("spark scale down"),
+ "executor-2" -> ExecutorDecommissionInfo("spark scale down")))
+ verify(client, never()).decommissionExecutors(any(), anyBoolean(), anyBoolean())
+ verify(client, never()).killExecutors(any(), anyBoolean(), anyBoolean(), anyBoolean())
+
+ // A stale timeout can be rejected again without accounting the executor as removed.
+ when(client.decommissionExecutorsIfIdle(any(), anyBoolean()))
+ .thenReturn(Seq.empty[String])
+ assert(removeExecutorsDefaultProfile(manager, Seq("executor-1")).isEmpty)
+ assert(executorsDecommissioning(manager) === Set("executor-2"))
+
+ // The same executor remains eligible for a later request once it is actually idle.
+ when(client.decommissionExecutorsIfIdle(any(), anyBoolean()))
+ .thenReturn(Seq("executor-1"))
+ assert(removeExecutorsDefaultProfile(manager, Seq("executor-1")) === Seq("executor-1"))
+ assert(executorsDecommissioning(manager) === Set("executor-1", "executor-2"))
+ }
+
+ test("SPARK-58879: unsupported idle decommission does not force removal") {
+ val unsupportedClient = mock(classOf[ExecutorAllocationClient], CALLS_REAL_METHODS)
+ assert(unsupportedClient.decommissionExecutorsIfIdle(
+ Array("executor-1" -> ExecutorDecommissionInfo("spark scale down")),
+ adjustTargetNumExecutors = false).isEmpty)
+ verify(unsupportedClient, never()).decommissionExecutors(any(), anyBoolean(), anyBoolean())
+ verify(unsupportedClient, never())
+ .killExecutors(any(), anyBoolean(), anyBoolean(), anyBoolean())
+ }
+
test("remove multiple executors") {
val manager = createManager(createConf(5, 10, 5))
(1 to 10).map(_.toString).foreach { id => onExecutorAddedDefaultProfile(manager, id) }
diff --git a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
index 8c317661ef8a..6d01593f3e1a 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.scheduler
import java.util.Properties
+import java.util.concurrent.{Callable, CountDownLatch, FutureTask, LinkedBlockingQueue, TimeUnit}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
import scala.collection.mutable
@@ -580,6 +581,153 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
}
}
+ Seq(false, true).foreach { isBarrier =>
+ test("SPARK-58879: idle decommission rejects tasks assigned before LaunchTask " +
+ s"(barrier=$isBarrier)") {
+ val numTasks = if (isBarrier) 2 else 1
+ val conf = new SparkConf().set(EXECUTOR_CORES, 2)
+ val backend = createDecommissionBackend(conf)
+ val scheduler = backend.taskScheduler
+ val executor = registerDecommissionExecutor(backend, "1", 2)
+ val decommissionCalled = new AtomicBoolean(false)
+ backend.beforeDecommission = (_, _, _) => decommissionCalled.set(true)
+ val launchEntered = new CountDownLatch(1)
+ val allowLaunch = new CountDownLatch(1)
+ executor.beforeLaunch = _ => {
+ launchEntered.countDown()
+ require(allowLaunch.await(30, TimeUnit.SECONDS), "LaunchTask was not released")
+ }
+ val taskSet = if (isBarrier) {
+ FakeTask.createBarrierTaskSet(numTasks)
+ } else {
+ FakeTask.createTaskSet(numTasks)
+ }
+ scheduler.submitTasks(taskSet)
+ backend.driverEndpoint.send(ReviveOffers)
+
+ var requestThread: Thread = null
+ try {
+ assert(launchEntered.await(10, TimeUnit.SECONDS))
+ assert(scheduler.runningTasksByExecutors("1") === numTasks)
+ val (thread, request) = startDecommissionRequest {
+ backend.decommissionExecutorsIfIdle(
+ Array("1" -> ExecutorDecommissionInfo("idle timeout")), false)
+ }
+ requestThread = thread
+ assert(request.get(10, TimeUnit.SECONDS).isEmpty)
+ assert(backend.isExecutorActive("1"))
+ assert(!executor.decommissionReceived)
+ assert(!decommissionCalled.get())
+ } finally {
+ allowLaunch.countDown()
+ if (requestThread != null) {
+ requestThread.join(TimeUnit.SECONDS.toMillis(10))
+ assert(!requestThread.isAlive)
+ }
+ }
+
+ val tasks = (0 until numTasks).map(_ => executor.nextTask())
+ flushDecommissionBackend(backend)
+ tasks.foreach(completeDecommissionTestTask(backend, _))
+ assert(!scheduler.isExecutorBusy("1"))
+ assert(backend.getExecutorAvailableCpus("1").contains(BigDecimal(2)))
+ assert(backend.decommissionExecutorsIfIdle(
+ Array("1" -> ExecutorDecommissionInfo("idle timeout")), false) === Seq("1"))
+ assert(executor.decommissionReceived)
+ }
+ }
+
+ test("SPARK-58879: idle decommission fences an executor before concurrent resource offers") {
+ val backend = createDecommissionBackend()
+ val retired = registerDecommissionExecutor(backend, "1")
+ val survivor = registerDecommissionExecutor(backend, "2")
+ backend.taskScheduler.submitTasks(FakeTask.createTaskSet(1))
+ val admissionEntered = new CountDownLatch(1)
+ val allowAdmission = new CountDownLatch(1)
+ val offerEntered = new CountDownLatch(1)
+ val locksHeld = new AtomicBoolean(false)
+ val decommissionReleased = new AtomicBoolean(false)
+ backend.beforeDecommission = (_, _, _) => {
+ locksHeld.set(Thread.holdsLock(backend.taskScheduler) && Thread.holdsLock(backend))
+ admissionEntered.countDown()
+ decommissionReleased.set(allowAdmission.await(30, TimeUnit.SECONDS))
+ }
+ backend.beforeOffers = () => offerEntered.countDown()
+
+ val (requestThread, request) = startDecommissionRequest {
+ backend.decommissionExecutorsIfIdle(
+ Array("1" -> ExecutorDecommissionInfo("idle timeout")), false)
+ }
+ try {
+ assert(admissionEntered.await(10, TimeUnit.SECONDS))
+ backend.driverEndpoint.send(ReviveOffers)
+ assert(offerEntered.await(10, TimeUnit.SECONDS))
+ } finally {
+ allowAdmission.countDown()
+ requestThread.join(TimeUnit.SECONDS.toMillis(10))
+ assert(!requestThread.isAlive)
+ }
+ assert(request.get(10, TimeUnit.SECONDS) === Seq("1"))
+ assert(locksHeld.get())
+ assert(decommissionReleased.get())
+ flushDecommissionBackend(backend)
+ assert(retired.launchedTasks.isEmpty)
+ assert(retired.decommissionReceived)
+ val task = survivor.nextTask()
+ assert(task.executorId === "2")
+ completeDecommissionTestTask(backend, task)
+ }
+
+ test("SPARK-58879: idle decommission skips duplicates and does not replay rejected requests") {
+ // Retention defaults to zero. Enable it so an incorrectly queued request is observable.
+ val conf = new SparkConf().set(SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS, 1)
+ val backend = createDecommissionBackend(conf)
+ val first = registerDecommissionExecutor(backend, "1")
+ val second = registerDecommissionExecutor(backend, "2")
+ val info = ExecutorDecommissionInfo("idle timeout")
+ val requests = mutable.ArrayBuffer.empty[(Seq[String], Boolean, Boolean)]
+ backend.beforeDecommission = (ids, adjustTarget, triggeredByExecutor) => {
+ requests += ((ids, adjustTarget, triggeredByExecutor))
+ }
+
+ assert(backend.decommissionExecutorsIfIdle(
+ Array("1" -> info, "1" -> info, "3" -> info), false) === Seq("1"))
+ assert(backend.decommissionExecutorsIfIdle(Array("1" -> info), false).isEmpty)
+ assert(requests.toSeq === Seq((Seq("1"), false, false)))
+ assert(first.decommissionReceived)
+ assert(!second.decommissionReceived)
+ assert(!backend.hasUnknownDecommission("1"))
+ assert(!backend.hasUnknownDecommission("3"))
+
+ val third = registerDecommissionExecutor(backend, "3")
+ assert(!third.decommissionReceived)
+ assert(backend.isExecutorActive("3"))
+
+ assert(backend.decommissionExecutors(Array("4" -> info), false, false).isEmpty)
+ assert(backend.hasUnknownDecommission("4"))
+ val fourth = registerDecommissionExecutor(backend, "4")
+ assert(fourth.decommissionReceived)
+ assert(!backend.isExecutorActive("4"))
+ assert(!backend.hasUnknownDecommission("4"))
+ }
+
+ test("SPARK-58879: forced decommission still accepts a busy executor") {
+ val backend = createDecommissionBackend()
+ val executor = registerDecommissionExecutor(backend, "1")
+ backend.taskScheduler.submitTasks(FakeTask.createTaskSet(1))
+ backend.driverEndpoint.send(ReviveOffers)
+ val task = executor.nextTask()
+ assert(backend.taskScheduler.isExecutorBusy("1"))
+
+ assert(backend.decommissionExecutors(
+ Array("1" -> ExecutorDecommissionInfo("host drain")), false, false) === Seq("1"))
+ assert(executor.decommissionReceived)
+ assert(!backend.isExecutorActive("1"))
+ completeDecommissionTestTask(backend, task)
+ assert(!backend.taskScheduler.isExecutorBusy("1"))
+ assert(executor.launchedTasks.isEmpty)
+ }
+
test("SPARK-41766: New registered executor should receive decommission request" +
" sent before registration") {
val conf = new SparkConf()
@@ -782,6 +930,59 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
assert(store.get().bytes === Array[Byte](50, 50))
}
+ private def createDecommissionBackend(
+ conf: SparkConf = new SparkConf()): DecommissionTestSchedulerBackend = {
+ conf.setMaster(s"coarseclustermanager[${classOf[DecommissionTestSchedulerBackend].getName}]")
+ .setAppName("idle decommission test")
+ .set(EXECUTOR_INSTANCES, 0)
+ .set(DYN_ALLOCATION_ENABLED, false)
+ .set(DECOMMISSION_ENABLED, true)
+ sc = new SparkContext(conf)
+ sc.schedulerBackend.asInstanceOf[DecommissionTestSchedulerBackend]
+ }
+
+ private def registerDecommissionExecutor(
+ backend: DecommissionTestSchedulerBackend,
+ executorId: String,
+ cores: Int = 1)
+ : DecommissionTestExecutorRpcEndpointRef = {
+ val executor = new DecommissionTestExecutorRpcEndpointRef(sc.conf, executorId)
+ assert(backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor(executorId, executor, "localhost", cores, Map.empty, Map.empty,
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)))
+ backend.driverEndpoint.send(LaunchedExecutor(executorId))
+ flushDecommissionBackend(backend)
+ executor
+ }
+
+ private def flushDecommissionBackend(backend: DecommissionTestSchedulerBackend): Unit = {
+ // Any synchronous request flushes earlier driver-endpoint messages. Ignore its result:
+ // executor "1" may not be registered or may already be retired.
+ backend.driverEndpoint.askSync[Boolean](IsExecutorAlive("1"))
+ }
+
+ private def completeDecommissionTestTask(
+ backend: DecommissionTestSchedulerBackend,
+ task: TaskDescription): Unit = {
+ val result = new DirectTaskResult[Int](
+ sc.env.serializer.newInstance().serialize(0), Seq.empty, Array.emptyLongArray)
+ val serializedResult = sc.env.closureSerializer.newInstance().serialize(result)
+ backend.driverEndpoint.send(StatusUpdate(
+ task.executorId, task.taskId, TaskState.FINISHED, new SerializableBuffer(serializedResult),
+ task.cpus, task.resources))
+ flushDecommissionBackend(backend)
+ }
+
+ private def startDecommissionRequest[T](body: => T): (Thread, FutureTask[T]) = {
+ val request = new FutureTask[T](new Callable[T] {
+ override def call(): T = body
+ })
+ val thread = new Thread(request, "idle-decommission-test")
+ thread.setDaemon(true)
+ thread.start()
+ (thread, request)
+ }
+
private def testSubmitJob(sc: SparkContext, rdd: RDD[Int]): Unit = {
sc.submitJob(
rdd,
@@ -793,7 +994,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
}
}
-/** Simple cluster manager that wires up our mock backend for the resource tests. */
+/** Cluster manager for the mock resource tests and real-scheduler decommission tests. */
private class CSMockExternalClusterManager extends ExternalClusterManager {
private var ts: TaskSchedulerImpl = _
@@ -804,12 +1005,18 @@ private class CSMockExternalClusterManager extends ExternalClusterManager {
override def createTaskScheduler(
sc: SparkContext,
masterURL: String): TaskScheduler = {
- ts = mock[TaskSchedulerImpl]
- when(ts.sc).thenReturn(sc)
- when(ts.applicationId()).thenReturn("appid1")
- when(ts.applicationAttemptId()).thenReturn(Some("attempt1"))
- when(ts.schedulingMode).thenReturn(SchedulingMode.FIFO)
- when(ts.excludedNodes()).thenReturn(Set.empty[String])
+ masterURL match {
+ case MOCK_REGEX(backendClassName)
+ if backendClassName == classOf[DecommissionTestSchedulerBackend].getName =>
+ ts = new TaskSchedulerImpl(sc, sc.conf.get(TASK_MAX_FAILURES))
+ case _ =>
+ ts = mock[TaskSchedulerImpl]
+ when(ts.sc).thenReturn(sc)
+ when(ts.applicationId()).thenReturn("appid1")
+ when(ts.applicationAttemptId()).thenReturn(Some("attempt1"))
+ when(ts.schedulingMode).thenReturn(SchedulingMode.FIFO)
+ when(ts.excludedNodes()).thenReturn(Set.empty[String])
+ }
ts
}
@@ -837,6 +1044,75 @@ class TestCoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, override v
def getTaskSchedulerImpl(): TaskSchedulerImpl = scheduler
}
+private[spark] class DecommissionTestSchedulerBackend(
+ scheduler: TaskSchedulerImpl,
+ override val rpcEnv: RpcEnv)
+ extends CoarseGrainedSchedulerBackend(scheduler, rpcEnv) {
+
+ val taskScheduler = scheduler
+ @volatile var beforeOffers: () => Unit = () => ()
+ @volatile var beforeDecommission: (Seq[String], Boolean, Boolean) => Unit = (_, _, _) => ()
+
+ // Tests drive the real offer paths explicitly, without periodic or scheduler-triggered offers.
+ override protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint {
+ override def onStart(): Unit = {}
+
+ override def receive: PartialFunction[Any, Unit] = {
+ case ReviveOffers =>
+ beforeOffers()
+ super.receive(ReviveOffers)
+ case message => super.receive(message)
+ }
+ }
+
+ override def reviveOffers(): Unit = {}
+
+ override def decommissionExecutors(
+ executorsAndDecomInfo: Array[(String, ExecutorDecommissionInfo)],
+ adjustTargetNumExecutors: Boolean,
+ triggeredByExecutor: Boolean): Seq[String] = {
+ beforeDecommission(
+ executorsAndDecomInfo.map(_._1).toSeq, adjustTargetNumExecutors, triggeredByExecutor)
+ super.decommissionExecutors(
+ executorsAndDecomInfo, adjustTargetNumExecutors, triggeredByExecutor)
+ }
+
+ def hasUnknownDecommission(executorId: String): Boolean = synchronized {
+ unknownExecutorsPendingDecommission.getIfPresent(executorId) != null
+ }
+}
+
+private[spark] class DecommissionTestExecutorRpcEndpointRef(
+ conf: SparkConf,
+ executorId: String) extends RpcEndpointRef(conf) {
+
+ val launchedTasks = new LinkedBlockingQueue[TaskDescription]()
+ @volatile var decommissionReceived = false
+ @volatile var beforeLaunch: TaskDescription => Unit = (_: TaskDescription) => ()
+
+ override def address: RpcAddress = RpcAddress("localhost", 10000 + executorId.toInt)
+ override def name: String = s"executor-$executorId"
+
+ override def send(message: Any): Unit = message match {
+ case LaunchTask(data) =>
+ val task = TaskDescription.decode(data.value)
+ beforeLaunch(task)
+ launchedTasks.add(task)
+ case DecommissionExecutor => decommissionReceived = true
+ case _ =>
+ }
+
+ override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T] = {
+ Future.successful(true.asInstanceOf[T])
+ }
+
+ def nextTask(): TaskDescription = {
+ val task = launchedTasks.poll(10, TimeUnit.SECONDS)
+ require(task != null, s"No task was launched on executor $executorId")
+ task
+ }
+}
+
private[spark] class MockExecutorRpcEndpointRef(conf: SparkConf) extends RpcEndpointRef(conf) {
// scalastyle:off executioncontextglobal
import scala.concurrent.ExecutionContext.Implicits.global
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
index 1f425d16edfb..5ac0d66f4183 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
@@ -27,24 +27,26 @@ import io.fabric8.kubernetes.client.dsl.PodResource
import io.fabric8.kubernetes.client.dsl.base.PatchContext
import org.jmock.lib.concurrent.DeterministicScheduler
import org.mockito.{ArgumentCaptor, Mock, MockitoAnnotations}
-import org.mockito.ArgumentMatchers.{any, eq => mockitoEq}
-import org.mockito.Mockito.{atLeastOnce, inOrder, mock, never, spy, verify, when}
+import org.mockito.ArgumentMatchers.{any, anyBoolean, eq => mockitoEq}
+import org.mockito.Mockito.{atLeastOnce, inOrder, mock, never, spy, times, verify, when}
import org.scalatest.BeforeAndAfter
import org.apache.spark.{SparkConf, SparkContext, SparkEnv, SparkException, SparkFunSuite}
import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.deploy.k8s.Fabric8Aliases._
+import org.apache.spark.internal.config.SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS
import org.apache.spark.resource.{ResourceProfile, ResourceProfileManager}
-import org.apache.spark.rpc.{RpcCallContext, RpcEndpoint, RpcEndpointRef, RpcEnv}
-import org.apache.spark.scheduler.{ExecutorKilled, ExecutorLossReason, LiveListenerBus, TaskSchedulerImpl}
-import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor, RemoveExecutor, StopDriver, StopExecutors}
+import org.apache.spark.rpc.{RpcAddress, RpcCallContext, RpcEndpoint, RpcEndpointRef, RpcEnv}
+import org.apache.spark.scheduler.{ExecutorDecommissionInfo, ExecutorKilled, ExecutorLossReason, LiveListenerBus, TaskSchedulerImpl}
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{DecommissionExecutor, RegisterExecutor, RemoveExecutor, StopDriver, StopExecutors}
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SPARK_APP_ID
+import org.apache.spark.storage.{BlockManager, BlockManagerMaster}
class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAndAfter {
- private val schedulerExecutorService = new DeterministicScheduler()
+ private var schedulerExecutorService: DeterministicScheduler = _
private val sparkConf = new SparkConf(false)
.set("spark.executor.instances", "3")
.set("spark.app.id", TEST_SPARK_APP_ID)
@@ -57,6 +59,12 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
@Mock
private var env: SparkEnv = _
+ @Mock
+ private var blockManager: BlockManager = _
+
+ @Mock
+ private var blockManagerMaster: BlockManagerMaster = _
+
@Mock
private var rpcEnv: RpcEnv = _
@@ -116,12 +124,17 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
private val defaultProfile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
before {
+ schedulerExecutorService = new DeterministicScheduler()
MockitoAnnotations.openMocks(this).close()
when(taskScheduler.sc).thenReturn(sc)
+ when(taskScheduler.excludedNodes()).thenReturn(Set.empty[String])
when(sc.conf).thenReturn(sparkConf)
+ when(sc.listenerBus).thenReturn(listenerBus)
when(sc.resourceProfileManager).thenReturn(resourceProfileManager)
when(sc.env).thenReturn(env)
when(env.rpcEnv).thenReturn(rpcEnv)
+ when(env.blockManager).thenReturn(blockManager)
+ when(blockManager.master).thenReturn(blockManagerMaster)
driverEndpoint = ArgumentCaptor.forClass(classOf[RpcEndpoint])
when(
rpcEnv.setupEndpoint(
@@ -134,7 +147,15 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
when(configMapsOperations.inNamespace("default")).thenReturn(configMapsWithNamespace)
when(configMapsWithNamespace.resource(any[ConfigMap]())).thenReturn(configMapResource)
when(podAllocator.driverPod).thenReturn(None)
- schedulerBackendUnderTest = new KubernetesClusterSchedulerBackend(
+ schedulerBackendUnderTest = createSchedulerBackend()
+ }
+
+ after {
+ ResourceProfile.clearDefaultProfile()
+ }
+
+ private def createSchedulerBackend(): KubernetesClusterSchedulerBackend = {
+ new KubernetesClusterSchedulerBackend(
taskScheduler,
sc,
kubernetesClient,
@@ -146,8 +167,35 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
pollEvents)
}
- after {
- ResourceProfile.clearDefaultProfile()
+ private def registerExecutor(
+ backend: KubernetesClusterSchedulerBackend,
+ executorId: String): RpcEndpointRef = {
+ val executorEndpoint = mock(classOf[RpcEndpointRef])
+ when(executorEndpoint.address).thenReturn(RpcAddress("localhost", 10000 + executorId.toInt))
+ backend.createDriverEndpoint().receiveAndReply(mock(classOf[RpcCallContext])).apply(
+ RegisterExecutor(executorId, executorEndpoint, s"host-$executorId", 1,
+ Map.empty, Map.empty, Map.empty, defaultProfile.id))
+ assert(backend.isExecutorActive(executorId))
+ executorEndpoint
+ }
+
+ private def withDecommissionMetadata(
+ f: KubernetesClusterSchedulerBackend => Unit): Unit = {
+ val keys = Seq(KUBERNETES_ALLOCATION_PODS_ALLOCATOR.key,
+ KUBERNETES_EXECUTOR_POD_DELETION_COST.key, SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS.key)
+ val originalValues = keys.map(key => key -> sparkConf.getOption(key))
+ sparkConf.set(KUBERNETES_ALLOCATION_PODS_ALLOCATOR, "deployment")
+ sparkConf.set(KUBERNETES_EXECUTOR_POD_DELETION_COST, 7)
+ sparkConf.set(SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS, 10)
+ try {
+ // The backend captures the unknown-executor cache size when it is constructed.
+ f(createSchedulerBackend())
+ } finally {
+ originalValues.foreach {
+ case (key, Some(value)) => sparkConf.set(key, value)
+ case (key, None) => sparkConf.remove(key)
+ }
+ }
}
test("Start all components") {
@@ -303,6 +351,83 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
sparkConf.remove(KUBERNETES_EXECUTOR_POD_DELETION_COST.key)
}
+ test("SPARK-58879: idle decommission passes only selected executors to Kubernetes") {
+ withDecommissionMetadata { schedulerBackend =>
+ val backend = spy[KubernetesClusterSchedulerBackend](schedulerBackend)
+ val idleExecutor = registerExecutor(backend, "1")
+ val busyExecutor = registerExecutor(backend, "2")
+ when(taskScheduler.isExecutorBusy("2")).thenReturn(true)
+ when(podsWithNamespace.withLabel(SPARK_APP_ID_LABEL, TEST_SPARK_APP_ID))
+ .thenReturn(labeledPods)
+ when(labeledPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE))
+ .thenReturn(labeledPods)
+ when(labeledPods.withLabelIn(SPARK_EXECUTOR_ID_LABEL, "1")).thenReturn(labeledPods)
+ val podResource = mock(classOf[PodResource])
+ when(labeledPods.resources())
+ .thenAnswer(_ => java.util.stream.Stream.of[PodResource](podResource))
+ val decomInfo = ExecutorDecommissionInfo("test")
+
+ val accepted = backend.decommissionExecutorsIfIdle(
+ Array("1" -> decomInfo, "2" -> decomInfo, "3" -> decomInfo, "1" -> decomInfo),
+ adjustTargetNumExecutors = false)
+
+ assert(accepted === Seq("1"))
+ val requests = ArgumentCaptor.forClass(
+ classOf[Array[(String, ExecutorDecommissionInfo)]])
+ verify(backend).decommissionExecutors(
+ requests.capture(), mockitoEq(false), mockitoEq(false))
+ assert(requests.getValue.toSeq === Seq("1" -> decomInfo))
+ assert(!backend.isExecutorActive("1"))
+ assert(backend.isExecutorActive("2"))
+ verify(blockManagerMaster).decommissionBlockManagers(Seq("1"))
+ verify(idleExecutor).send(DecommissionExecutor)
+ verify(busyExecutor, never()).send(DecommissionExecutor)
+ verify(kubernetesClient, never()).pods()
+
+ schedulerExecutorService.runUntilIdle()
+ verify(labeledPods, times(2)).withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+ val executorIds = ArgumentCaptor.forClass(classOf[Array[String]])
+ verify(labeledPods, atLeastOnce()).withLabelIn(
+ mockitoEq(SPARK_EXECUTOR_ID_LABEL), executorIds.capture(): _*)
+ assert(executorIds.getAllValues.asScala.forall(_.toSeq == Seq("1")))
+ verify(labeledPods, times(2)).resources()
+ val patches = ArgumentCaptor.forClass(classOf[Pod])
+ verify(podResource, times(2)).patch(any(classOf[PatchContext]), patches.capture())
+ val appliedPods = patches.getAllValues.asScala
+ assert(appliedPods.exists { pod =>
+ Option(pod.getMetadata.getLabels).exists(_.get("soLong") == "cruelWorld")
+ })
+ assert(appliedPods.exists { pod =>
+ Option(pod.getMetadata.getAnnotations).exists(_.get(POD_DELETION_COST) == "7")
+ })
+ }
+ }
+
+ test("SPARK-58879: rejected idle decommission has no pod updates or replay") {
+ withDecommissionMetadata { schedulerBackend =>
+ val backend = spy[KubernetesClusterSchedulerBackend](schedulerBackend)
+ val busyExecutor = registerExecutor(backend, "1")
+ when(taskScheduler.isExecutorBusy("1")).thenReturn(true)
+ val decomInfo = ExecutorDecommissionInfo("test")
+
+ assert(backend.decommissionExecutorsIfIdle(
+ Array.empty[(String, ExecutorDecommissionInfo)],
+ adjustTargetNumExecutors = false).isEmpty)
+ assert(backend.decommissionExecutorsIfIdle(
+ Array("1" -> decomInfo, "2" -> decomInfo, "1" -> decomInfo),
+ adjustTargetNumExecutors = false).isEmpty)
+
+ val laterExecutor = registerExecutor(backend, "2")
+ schedulerExecutorService.runUntilIdle()
+ verify(backend, never()).decommissionExecutors(
+ any[Array[(String, ExecutorDecommissionInfo)]](), anyBoolean(), anyBoolean())
+ verify(blockManagerMaster, never()).decommissionBlockManagers(any[Seq[String]]())
+ verify(busyExecutor, never()).send(DecommissionExecutor)
+ verify(laterExecutor, never()).send(DecommissionExecutor)
+ verify(kubernetesClient, never()).pods()
+ }
+ }
+
test("SPARK-34407: CoarseGrainedSchedulerBackend.stop may throw SparkException") {
schedulerBackendUnderTest.start()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org