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 <[email protected]>
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 <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
(cherry picked from commit b1ce1292b62601a12a5203bbcd9e09fad49a6cfc)
Signed-off-by: Chao Sun <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
