This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 924ebbb6619 [SPARK-41766][CORE] Handle decommission request sent
before executor registration
924ebbb6619 is described below
commit 924ebbb6619af784dcb42a27f2b6e36fd1523e1d
Author: Warren Zhu <warren.zh...@gmail.com>
AuthorDate: Mon May 1 15:03:28 2023 -0700
[SPARK-41766][CORE] Handle decommission request sent before executor
registration
### What changes were proposed in this pull request?
Handle decommission request sent before executor registration. When decom
request is triggered from worker, executor should be decommissioned and not
accept new tasks, but if scheduler backend got decom request before executor
registration, the executor won't be decommissioned. So cache the unknown
executors and resent decommission request after executor registration.
### Why are the changes needed?
Current behavior is such requests will be ignored
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added test in CoarseGrainedSchedulerBackendSuite
Closes #39280 from warrenzhu25/decom-missing.
Authored-by: Warren Zhu <warren.zh...@gmail.com>
Signed-off-by: Dongjoon Hyun <dongj...@apache.org>
---
.../org/apache/spark/internal/config/package.scala | 10 +++++
.../cluster/CoarseGrainedSchedulerBackend.scala | 24 ++++++++++
.../CoarseGrainedSchedulerBackendSuite.scala | 51 ++++++++++++++++++++--
3 files changed, 82 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala
b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 125025d6a2f..fa88606e1f6 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -2260,6 +2260,16 @@ package object config {
.checkValue(_ >= 0, "needs to be a non-negative value")
.createWithDefault(0)
+ private[spark] val SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS =
+ ConfigBuilder("spark.scheduler.maxRetainedUnknownDecommissionExecutors")
+ .internal()
+ .doc("Max number of unknown executors by decommission to retain. This
affects " +
+ "whether executor could receive decommission request sent before its
registration.")
+ .version("3.5.0")
+ .intConf
+ .checkValue(_ >= 0, "needs to be a non-negative value")
+ .createWithDefault(0)
+
private[spark] val PUSH_BASED_SHUFFLE_ENABLED =
ConfigBuilder("spark.shuffle.push.enabled")
.doc("Set to true to enable push-based shuffle on the client side and
this works in " +
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 1be20bef012..aeac2616711 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
@@ -24,6 +24,7 @@ import javax.annotation.concurrent.GuardedBy
import scala.collection.mutable.{HashMap, HashSet, Queue}
import scala.concurrent.Future
+import com.google.common.cache.CacheBuilder
import org.apache.hadoop.security.UserGroupInformation
import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState}
@@ -102,6 +103,15 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
// Executors which are being decommissioned. Maps from executorId to
ExecutorDecommissionInfo.
protected val executorsPendingDecommission = new HashMap[String,
ExecutorDecommissionInfo]
+ // Unknown Executors which are being decommissioned. This could be caused by
unregistered executor
+ // This executor should be decommissioned after registration.
+ // Maps from executorId to (ExecutorDecommissionInfo,
adjustTargetNumExecutors,
+ // triggeredByExecutor).
+ protected val unknownExecutorsPendingDecommission =
+ CacheBuilder.newBuilder()
+ .maximumSize(conf.get(SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS))
+ .build[String, (ExecutorDecommissionInfo, Boolean, Boolean)]()
+
// A map of ResourceProfile id to map of hostname with its possible task
number running on it
@GuardedBy("CoarseGrainedSchedulerBackend.this")
protected var rpHostToLocalTaskCount: Map[Int, Map[String, Int]] = Map.empty
@@ -295,6 +305,12 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
listenerBus.post(
SparkListenerExecutorAdded(System.currentTimeMillis(), executorId,
data))
// Note: some tests expect the reply to come after we put the
executor in the map
+ // Decommission executor whose request received before registration
+ Option(unknownExecutorsPendingDecommission.getIfPresent(executorId))
+ .foreach(v => {
+ decommissionExecutors(Array((executorId, v._1)), v._2, v._3)
+ unknownExecutorsPendingDecommission.invalidate(executorId)
+ })
context.reply(true)
}
@@ -528,14 +544,22 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
// Do not change this code without running the K8s integration suites
val executorsToDecommission = executorsAndDecomInfo.flatMap { case
(executorId, decomInfo) =>
// Only bother decommissioning executors which are alive.
+ // Keep executor decommission info in case executor started, but not
registered yet
if (isExecutorActive(executorId)) {
scheduler.executorDecommission(executorId, decomInfo)
executorsPendingDecommission(executorId) = decomInfo
Some(executorId)
} else {
+ unknownExecutorsPendingDecommission.put(executorId,
+ (decomInfo, adjustTargetNumExecutors, triggeredByExecutor))
None
}
}
+
+ if (executorsToDecommission.isEmpty) {
+ return executorsToDecommission
+ }
+
logInfo(s"Decommission executors: ${executorsToDecommission.mkString(",
")}")
// If we don't want to replace the executors we are decommissioning
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 15c8ab16ec9..c4d4fd7d80e 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -21,8 +21,10 @@ import java.util.Properties
import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
+import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps
+import scala.reflect.ClassTag
import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito.when
@@ -37,10 +39,9 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.resource.{ExecutorResourceRequests,
ResourceInformation, ResourceProfile, TaskResourceRequests}
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.resource.TestResourceIDs._
-import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcEnv}
+import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcEnv, RpcTimeout}
+import org.apache.spark.scheduler.cluster.{CoarseGrainedSchedulerBackend,
ExecutorInfo}
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
-import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
-import org.apache.spark.scheduler.cluster.ExecutorInfo
import org.apache.spark.util.{RpcUtils, SerializableBuffer, Utils}
class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with
LocalSparkContext
@@ -492,6 +493,32 @@ class CoarseGrainedSchedulerBackendSuite extends
SparkFunSuite with LocalSparkCo
}
}
+ test("SPARK-41766: New registered executor should receive decommission
request" +
+ " sent before registration") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+ .set(SCHEDULER_MAX_RETAINED_UNKNOWN_EXECUTORS.key, "1")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val mockEndpointRef = new MockExecutorRpcEndpointRef(conf)
+ val mockAddress = mock[RpcAddress]
+ val executorId = "1"
+ val executorDecommissionInfo = ExecutorDecommissionInfo(
+ s"Executor $executorId is decommissioned")
+
+ backend.decommissionExecutor(executorId, executorDecommissionInfo, false)
+ assert(!mockEndpointRef.decommissionReceived)
+
+ backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map(), Map(),
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+
+ sc.listenerBus.waitUntilEmpty(executorUpTimeout.toMillis)
+ assert(mockEndpointRef.decommissionReceived)
+ }
+
private def testSubmitJob(sc: SparkContext, rdd: RDD[Int]): Unit = {
sc.submitJob(
rdd,
@@ -546,3 +573,21 @@ class TestCoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, override v
def getTaskSchedulerImpl(): TaskSchedulerImpl = scheduler
}
+
+private[spark] class MockExecutorRpcEndpointRef(conf: SparkConf) extends
RpcEndpointRef(conf) {
+ // scalastyle:off executioncontextglobal
+ import scala.concurrent.ExecutionContext.Implicits.global
+ // scalastyle:on executioncontextglobal
+
+ var decommissionReceived = false
+
+ override def address: RpcAddress = null
+ override def name: String = "executor"
+ override def send(message: Any): Unit =
+ message match {
+ case DecommissionExecutor => decommissionReceived = true
+ }
+ override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]
= {
+ Future{true.asInstanceOf[T]}
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
