[SPARK-12281][Core]Fix a race condition when reporting ExecutorState in the shutdown hook

core/src/main/scala/org/apache/spark/deploy/LocalSparkCluster.scala

@@ -75,6 +75,8 @@ class LocalSparkCluster(
     // Stop the workers before the master so they don't get upset that it disconnected
     workerRpcEnvs.foreach(_.shutdown())
     masterRpcEnvs.foreach(_.shutdown())
+    workerRpcEnvs.foreach(_.awaitTermination())
+    masterRpcEnvs.foreach(_.awaitTermination())
     masterRpcEnvs.clear()
     workerRpcEnvs.clear()
   }

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -257,9 +257,8 @@ private[deploy] class Master(
           exec.state = state
 
           if (state == ExecutorState.RUNNING) {
-            if (oldState != ExecutorState.LAUNCHING) {
-              logWarning(s"Executor $execId state transfer from $oldState to RUNNING is unexpected")
-            }
+            assert(oldState == ExecutorState.LAUNCHING,
+              s"executor $execId state transfer from $oldState to RUNNING is illegal")
             appInfo.resetRetryCount()
           }
 

core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala

@@ -71,6 +71,11 @@ private[deploy] class ExecutorRunner(
     workerThread.start()
     // Shutdown hook that kills actors on shutdown.
     shutdownHook = ShutdownHookManager.addShutdownHook { () =>
+      // It's possible that we arrive here before calling `fetchAndRunExecutor`, then `state` will
+      // be `ExecutorState.RUNNING`. In this case, we should set `state` to `FAILED`.
+      if (state == ExecutorState.RUNNING) {
+        state = ExecutorState.FAILED
+      }
       killProcess(Some("Worker shutting down")) }
   }