CR feedback/cleanup

Author: holdenk

core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala

@@ -87,15 +87,15 @@ private[spark] trait ExecutorAllocationClient {
    * Default implementation delegates to kill, scheduler must override
    * if it supports graceful decommissioning.
    *
-   * @param executors identifiers of executors & decom info.
+   * @param executorsAndDecominfo identifiers of executors & decom info.
    * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down
    *                                 after these executors have been decommissioned.
    * @return the ids of the executors acknowledged by the cluster manager to be removed.
    */
   def decommissionExecutors(
-    executors: Seq[(String, ExecutorDecommissionInfo)],
+    executorsAndDecomInfo: Seq[(String, ExecutorDecommissionInfo)],
     adjustTargetNumExecutors: Boolean): Seq[String] = {
-    killExecutors(executors.map(_._1),
+    killExecutors(executorsAndDecomInfo.map(_._1),
       adjustTargetNumExecutors,
       countFailures = false)
   }
@@ -108,13 +108,15 @@ private[spark] trait ExecutorAllocationClient {
    *
    * @param executorId identifiers of executor to decommission
    * @param decommissionInfo information about the decommission (reason, host loss)
+   * @param adjustTargetNumExecutors if we should adjust the target number of executors.
    * @return whether the request is acknowledged by the cluster manager.
    */
   def decommissionExecutor(executorId: String,
-      decommissionInfo: ExecutorDecommissionInfo): Boolean = {
+      decommissionInfo: ExecutorDecommissionInfo,
+      adjustTargetNumExecutors: Boolean): Boolean = {
     val decommissionedExecutors = decommissionExecutors(
       Seq((executorId, decommissionInfo)),
-      adjustTargetNumExecutors = true)
+      adjustTargetNumExecutors = adjustTargetNumExecutors)
     decommissionedExecutors.nonEmpty && decommissionedExecutors(0).equals(executorId)
   }
 

core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

@@ -546,7 +546,7 @@ private[spark] class ExecutorAllocationManager(
         val newExecutorTotal = numExecutorsTotalPerRpId.getOrElseUpdate(rpId,
           (executorMonitor.executorCountWithResourceProfile(rpId) -
             executorMonitor.pendingRemovalCountPerResourceProfileId(rpId) -
-            executorMonitor.pendingDecommissioningPerResourceProfileId(rpId)
+            executorMonitor.decommissioningPerResourceProfileId(rpId)
           ))
         if (newExecutorTotal - 1 < minNumExecutors) {
           logDebug(s"Not removing idle executor $executorIdToBeRemoved because there " +

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

@@ -193,8 +193,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
       case DecommissionExecutor(executorId, decommissionInfo) =>
         logError(s"Received decommission executor message ${executorId}: $decommissionInfo")
-        decommissionExecutors(Seq((executorId, decommissionInfo)),
-          adjustTargetNumExecutors = false)
+        decommissionExecutor(executorId, decommissionInfo)
 
       case RemoveWorker(workerId, host, message) =>
         removeWorker(workerId, host, message)
@@ -275,8 +274,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
       case DecommissionExecutor(executorId, decommissionInfo) =>
         logError(s"Received decommission executor message ${executorId}: ${decommissionInfo}.")
-        decommissionExecutors(Seq((executorId, decommissionInfo)),
-          adjustTargetNumExecutors = false)
+        decommissionExecutor(executorId, decommissionInfo)
         context.reply(true)
 
       case RetrieveSparkAppConfig(resourceProfileId) =>
@@ -423,6 +421,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
     /**
      * Mark a given executor as decommissioned and stop making resource offers for it.
+     *
      */
     private def decommissionExecutor(
         executorId: String, decommissionInfo: ExecutorDecommissionInfo): Boolean = {
@@ -508,16 +507,16 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   /**
    * Request that the cluster manager decommission the specified executors.
    *
-   * @param executors Identifiers of executors & decommission info.
+   * @param executorsAndDecomInfo Identifiers of executors & decommission info.
    * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down
    *                                 after these executors have been decommissioned.
    * @return the ids of the executors acknowledged by the cluster manager to be removed.
    */
   override def decommissionExecutors(
-      executors: Seq[(String, ExecutorDecommissionInfo)],
+      executorsAndDecomInfo: Seq[(String, ExecutorDecommissionInfo)],
       adjustTargetNumExecutors: Boolean): Seq[String] = {
 
-    val executorsToDecommission = executors.filter{case (executorId, _) =>
+    val executorsToDecommission = executorsAndDecomInfo.filter { case (executorId, _) =>
       CoarseGrainedSchedulerBackend.this.synchronized {
         // Only bother decommissioning executors which are alive.
         if (isExecutorActive(executorId)) {
@@ -532,7 +531,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     // If we don't want to replace the executors we are decommissioning
     if (adjustTargetNumExecutors) {
       executorsToDecommission.foreach { case (exec, _) =>
-        val rpId = executorDataMap(exec).resourceProfileId
+        val rpId = withLock {
+          executorDataMap(exec).resourceProfileId
+        }
         val rp = scheduler.sc.resourceProfileManager.resourceProfileFromId(rpId)
         if (requestedTotalExecutorsPerResourceProfile.isEmpty) {
           // Assume that we are killing an executor that was started by default and
@@ -546,7 +547,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
       doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
     }
 
-    val decommissioned = executorsToDecommission.filter{case (executorId, decomInfo) =>
+    val decommissioned = executorsToDecommission.filter { case (executorId, decomInfo) =>
       doDecommission(executorId, decomInfo)
     }.map(_._1)
     decommissioned
@@ -556,7 +557,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   private def doDecommission(executorId: String,
       decomInfo: ExecutorDecommissionInfo): Boolean = {
 
-    logInfo(s"Starting decommissioning executor $executorId.")
+    logInfo(s"Asking executor $executorId to decommissioning.")
     try {
       scheduler.executorDecommission(executorId, decomInfo)
       if (driverEndpoint != null) {
@@ -580,12 +581,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
           return false
       }
     }
-    logInfo(s"Finished decommissioning executor $executorId.")
+    logInfo(s"Asked executor $executorId to decommission.")
 
     if (conf.get(STORAGE_DECOMMISSION_ENABLED)) {
       try {
-        logInfo("Starting decommissioning block manager corresponding to " +
-          s"executor $executorId.")
+        logInfo(s"Asking block manager corresponding to executor $executorId to decommission.")
         scheduler.sc.env.blockManager.master.decommissionBlockManagers(Seq(executorId))
       } catch {
         case e: Exception =>
@@ -595,6 +595,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
       }
       logInfo(s"Acknowledged decommissioning block manager corresponding to $executorId.")
     }
+
     true
   }
 

core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

@@ -115,7 +115,7 @@ private[spark] class ExecutorMonitor(
       var newNextTimeout = Long.MaxValue
       timedOutExecs = executors.asScala
         .filter { case (_, exec) =>
-          !exec.pendingRemoval && !exec.hasActiveShuffle && !exec.pendingDecommissioning}
+          !exec.pendingRemoval && !exec.hasActiveShuffle && !exec.decommissioning}
         .filter { case (_, exec) =>
           val deadline = exec.timeoutAt
           if (deadline > now) {
@@ -155,7 +155,7 @@ private[spark] class ExecutorMonitor(
     ids.foreach { id =>
       val tracker = executors.get(id)
       if (tracker != null) {
-        tracker.pendingDecommissioning = true
+        tracker.decommissioning = true
       }
     }
 
@@ -186,13 +186,13 @@ private[spark] class ExecutorMonitor(
     executors.asScala.filter { case (k, v) => v.resourceProfileId == id && v.pendingRemoval }.size
   }
 
-  def pendingDecommissioningCount: Int = executors.asScala.count { case (_, exec) =>
-    exec.pendingDecommissioning
+  def decommissioningCount: Int = executors.asScala.count { case (_, exec) =>
+    exec.decommissioning
   }
 
-  def pendingDecommissioningPerResourceProfileId(id: Int): Int = {
+  def decommissioningPerResourceProfileId(id: Int): Int = {
     executors.asScala.filter { case (k, v) =>
-      v.resourceProfileId == id && v.pendingDecommissioning
+      v.resourceProfileId == id && v.decommissioning
     }.size
   }
 
@@ -352,7 +352,7 @@ private[spark] class ExecutorMonitor(
     val removed = executors.remove(event.executorId)
     if (removed != null) {
       decrementExecResourceProfileCount(removed.resourceProfileId)
-      if (!removed.pendingRemoval || !removed.pendingDecommissioning) {
+      if (!removed.pendingRemoval || !removed.decommissioning) {
         nextTimeout.set(Long.MinValue)
       }
     }
@@ -457,7 +457,7 @@ private[spark] class ExecutorMonitor(
 
   // Visible for testing
   private[spark] def executorsDecommissioning(): Set[String] = {
-    executors.asScala.filter { case (_, exec) => exec.pendingDecommissioning }.keys.toSet
+    executors.asScala.filter { case (_, exec) => exec.decommissioning }.keys.toSet
   }
 
   /**
@@ -512,7 +512,7 @@ private[spark] class ExecutorMonitor(
     @volatile var timedOut: Boolean = false
 
     var pendingRemoval: Boolean = false
-    var pendingDecommissioning: Boolean = false
+    var decommissioning: Boolean = false
     var hasActiveShuffle: Boolean = false
 
     private var idleStart: Long = -1

core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala

@@ -314,13 +314,9 @@ private[storage] class BlockManagerDecommissioner(
     logInfo("Starting block migration thread")
     if (conf.get(config.STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED)) {
       rddBlockMigrationExecutor.submit(rddBlockMigrationRunnable)
-    } else {
-      stoppedRDD = true
     }
     if (conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED)) {
       shuffleBlockMigrationRefreshExecutor.submit(shuffleBlockMigrationRefreshRunnable)
-    } else {
-      stoppedShuffle = true
     }
     if (!conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED) &&
       !conf.get(config.STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED)) {

core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala

@@ -48,7 +48,6 @@ class BlockManagerMaster(
    */
   def decommissionBlockManagers(executorIds: Seq[String]): Unit = {
     driverEndpoint.ask[Boolean](DecommissionBlockManagers(executorIds))
-    logInfo(s"Decommissioning block managers on ${executorIds}")
   }
 
   /** Get Replication Info for all the RDD blocks stored in given blockManagerId */

core/src/test/scala/org/apache/spark/scheduler/WorkerDecommissionExtendedSuite.scala

@@ -65,7 +65,8 @@ class WorkerDecommissionExtendedSuite extends SparkFunSuite with LocalSparkConte
 
       val sched = sc.schedulerBackend.asInstanceOf[StandaloneSchedulerBackend]
       sc.getExecutorIds().tail.foreach { id =>
-        sched.decommissionExecutor(id, ExecutorDecommissionInfo("", false))
+        sched.decommissionExecutor(id, ExecutorDecommissionInfo("", false),
+          adjustTargetNumExecutors = false)
         assert(rdd3.sortByKey().collect().length === 100)
       }
     }

core/src/test/scala/org/apache/spark/scheduler/WorkerDecommissionSuite.scala

@@ -76,7 +76,9 @@ class WorkerDecommissionSuite extends SparkFunSuite with LocalSparkContext {
     // decom.sh message passing is tested manually.
     val sched = sc.schedulerBackend.asInstanceOf[StandaloneSchedulerBackend]
     val execs = sched.getExecutorIds()
-    execs.foreach(execId => sched.decommissionExecutor(execId, ExecutorDecommissionInfo("", false)))
+    // Make the executors decommission, finish, exit, and not be replaced.
+    val execsAndDecomInfo = execs.map((_, ExecutorDecommissionInfo("", false)))
+    sched.decommissionExecutors(execsAndDecomInfo, adjustTargetNumExecutors = true)
     val asyncCountResult = ThreadUtils.awaitResult(asyncCount, 20.seconds)
     assert(asyncCountResult === 10)
   }

core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionIntegrationSuite.scala

@@ -190,7 +190,8 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
     logInfo(s"Decommissioning executor ${execToDecommission}")
     sched.decommissionExecutor(
       execToDecommission,
-      ExecutorDecommissionInfo("", isHostDecommissioned = false))
+      ExecutorDecommissionInfo("", isHostDecommissioned = false),
+      adjustTargetNumExecutors = true)
     val decomTime = new SystemClock().getTimeMillis()
 
     // Wait for job to finish.

core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala

@@ -221,7 +221,5 @@ class BlockManagerDecommissionUnitSuite extends SparkFunSuite with Matchers {
     } finally {
         bmDecomManager.stop()
     }
-
-    bmDecomManager.stop()
   }
 }