Allow tasks to launch in the executor base class as well

Author: holdenk

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

@@ -248,12 +248,11 @@ private[spark] class Executor(
   }
 
   def launchTask(context: ExecutorBackend, taskDescription: TaskDescription): Unit = {
-    if (!decommissioned) {
-      val tr = new TaskRunner(context, taskDescription)
-      runningTasks.put(taskDescription.taskId, tr)
-      threadPool.execute(tr)
-    } else {
-      log.info(s"Not launching task, executor is in decommissioned state")
+    val tr = new TaskRunner(context, taskDescription)
+    runningTasks.put(taskDescription.taskId, tr)
+    threadPool.execute(tr)
+    if (decommissioned) {
+      log.error(s"Launching a task while in decommissioned state.")
     }
   }
 

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

@@ -428,7 +428,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
         scheduler.executorDecommission(executorId)
       }
       shouldDisable
-  }
+    }
 
     /**
      * Stop making resource offers for the given executor. The executor is marked as lost with