Code review feedback, various cleanups

Author: holdenk

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -79,7 +79,6 @@ private[spark] class CoarseGrainedExecutorBackend(
    */
   private[executor] val taskResources = new mutable.HashMap[Long, Map[String, ResourceInformation]]
 
-  // Track our decommissioning status internally.
   @volatile private var decommissioned = false
 
   override def onStart(): Unit = {
@@ -291,7 +290,7 @@ private[spark] class CoarseGrainedExecutorBackend(
       // is viewed as acceptable to minimize introduction of any new locking structures in critical
       // code paths.
 
-      val shutdownThread = new Thread() {
+      val shutdownThread = ThreadUtils.runInNewThread("wait-for-blocks-to-migrate") {
         var lastTaskRunningTime = System.nanoTime()
         val sleep_time = 1000 // 1s
 
@@ -300,11 +299,10 @@ private[spark] class CoarseGrainedExecutorBackend(
           if (executor == null || executor.numRunningTasks == 0) {
             if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
               logInfo("No running tasks, checking migrations")
-              val allBlocksMigrated = env.blockManager.lastMigrationInfo()
+              val (migrationTime, allBlocksMigrated) = env.blockManager.lastMigrationInfo()
               // We can only trust allBlocksMigrated boolean value if there were no tasks running
               // since the start of computing it.
-              if (allBlocksMigrated._2 &&
-                (allBlocksMigrated._1 > lastTaskRunningTime)) {
+              if (allBlocksMigrated && (migrationTime > lastTaskRunningTime)) {
                 logInfo("No running tasks, all blocks migrated, stopping.")
                 exitExecutor(0, "Finished decommissioning", notifyDriver = true)
               } else {
@@ -316,7 +314,7 @@ private[spark] class CoarseGrainedExecutorBackend(
             }
             Thread.sleep(sleep_time)
           } else {
-            logInfo("Blocked from shutdown by running task")
+            logInfo("Blocked from shutdown by running ${executor.numRunningtasks} tasks")
             // If there is a running task it could store blocks, so make sure we wait for a
             // migration loop to complete after the last task is done.
             Thread.sleep(sleep_time)

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

@@ -137,6 +137,6 @@ private[spark] object CoarseGrainedClusterMessages {
   // The message to check if `CoarseGrainedSchedulerBackend` thinks the executor is alive or not.
   case class IsExecutorAlive(executorId: String) extends CoarseGrainedClusterMessage
 
-  // Used to ask an executor to decommission it's self.
+  // Used to ask an executor to decommission itself. (Can be an internal message)
   case object DecommissionSelf extends CoarseGrainedClusterMessage
 }

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

@@ -442,8 +442,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
           case e: Exception =>
             logError(s"Unexpected error during decommissioning ${e.toString}", e)
         }
-        // Send decommission message to the executor (it could have originated on the executor
-        // but not necessarily.
+        // Send decommission message to the executor, this may be a duplicate since the executor
+        // could have been the one to notify us. But it's also possible the notification came from
+        // elsewhere and the executor does not yet know.
         executorDataMap.get(executorId) match {
           case Some(executorInfo) =>
             executorInfo.executorEndpoint.send(DecommissionSelf)

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

@@ -1823,7 +1823,7 @@ private[spark] class BlockManager(
   }
 
   /*
-   *  Returns the last migration time and a boolean for if all blocks have been migrated.
+   *  Returns the last migration time and a boolean denoting if all the blocks have been migrated.
    *  If there are any tasks running since that time the boolean may be incorrect.
    */
   private[spark] def lastMigrationInfo(): (Long, Boolean) = {

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

@@ -102,7 +102,6 @@ private[storage] class BlockManagerDecommissioner(
               } else {
                 logError(s"Skipping block ${shuffleBlockInfo} because it has failed ${retryCount}")
               }
-              logInfo(s"Migrated ${shuffleBlockInfo}")
               numMigratedShuffles.incrementAndGet()
           }
         }
@@ -135,8 +134,10 @@ private[storage] class BlockManagerDecommissioner(
 
   // Set if we encounter an error attempting to migrate and stop.
   @volatile private var stopped = false
-  @volatile private var stoppedRDD = false
-  @volatile private var stoppedShuffle = false
+  @volatile private var stoppedRDD =
+    !conf.get(config.STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED)
+  @volatile private var stoppedShuffle =
+    !conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED)
 
   private val migrationPeers =
     mutable.HashMap[BlockManagerId, ShuffleMigrationRunnable]()
@@ -301,13 +302,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/test/scala/org/apache/spark/storage/BlockManagerDecommissionUnitSuite.scala

@@ -38,7 +38,7 @@ class BlockManagerDecommissionUnitSuite extends SparkFunSuite with Matchers {
   private val sparkConf = new SparkConf(false)
     .set(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED, true)
     .set(config.STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED, true)
-    // Just replicate blocks as fast as we can during testing, there isn't another
+    // Just replicate blocks quickly during testing, as there isn't another
     // workload we need to worry about.
     .set(config.STORAGE_DECOMMISSION_REPLICATION_REATTEMPT_INTERVAL, 10L)
 
@@ -92,6 +92,5 @@ class BlockManagerDecommissionUnitSuite extends SparkFunSuite with Matchers {
     } finally {
         bmDecomManager.stop()
     }
-    bmDecomManager.stop()
   }
 }