Use decommissionExecutors and set adjustTargetNumExecutors to false so that we can match the pattern for killExecutor/killExecutors

Author: holdenk

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -193,7 +193,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
       case DecommissionExecutor(executorId) =>
         logError(s"Received decommission executor message ${executorId}.")
-        decommissionExecutor(executorId)
+        decommissionExecutors(Seq(executorId), adjustTargetNumExecutors = false)
 
       case RemoveWorker(workerId, host, message) =>
         removeWorker(workerId, host, message)
@@ -274,7 +274,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
       case DecommissionExecutor(executorId) =>
         logError(s"Received decommission executor message ${executorId}.")
-        decommissionExecutor(executorId)
+        decommissionExecutors(Seq(executorId), adjustTargetNumExecutors = false)
         context.reply(true)
 
       case RetrieveSparkAppConfig(resourceProfileId) =>

core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala

@@ -176,7 +176,7 @@ private[spark] class StandaloneSchedulerBackend(
 
   override def executorDecommissioned(fullId: String, message: String) {
     logInfo("Asked to decommission executor")
-    decommissionExecutor(fullId.split("/")(1))
+    decommissionExecutors(Seq(fullId.split("/")(1)), adjustTargetNumExecutors = false)
     logInfo("Executor %s decommissioned: %s".format(fullId, message))
   }