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

[zsxwing]

1. Make sure workers and masters exit so that no worker or master will still be running when triggering the shutdown hook.
2. Set ExecutorState to FAILED if it's still RUNNING when executing the shutdown hook.

This should fix the potential exceptions when exiting a local cluster

java.lang.AssertionError: assertion failed: executor 4 state transfer from RUNNING to RUNNING is illegal
    at scala.Predef$.assert(Predef.scala:179)
    at org.apache.spark.deploy.master.Master$$anonfun$receive$1.applyOrElse(Master.scala:260)
    at org.apache.spark.rpc.netty.Inbox$$anonfun$process$1.apply$mcV$sp(Inbox.scala:116)
    at org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:204)
    at org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:100)
    at org.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:215)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

java.lang.IllegalStateException: Shutdown hooks cannot be modified during shutdown.
    at org.apache.spark.util.SparkShutdownHookManager.add(ShutdownHookManager.scala:246)
    at org.apache.spark.util.ShutdownHookManager$.addShutdownHook(ShutdownHookManager.scala:191)
    at org.apache.spark.util.ShutdownHookManager$.addShutdownHook(ShutdownHookManager.scala:180)
    at org.apache.spark.deploy.worker.ExecutorRunner.start(ExecutorRunner.scala:73)
    at org.apache.spark.deploy.worker.Worker$$anonfun$receive$1.applyOrElse(Worker.scala:474)
    at org.apache.spark.rpc.netty.Inbox$$anonfun$process$1.apply$mcV$sp(Inbox.scala:116)
    at org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:204)
    at org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:100)
    at org.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:215)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

[andrewor14]

By the way, it seems that you are using an old version of Spark because we made the assertion error a warning in #10091. However, it seems that the RPC event loop already catches all non-fatal exceptions so it's actually safe to throw the assertion error if things are broken. Since this patch also fixes the error I think we should just revert #10091 here.

[andrewor14]

Once you do that this patch LGTM

[zsxwing]

retest this please

[SparkQA]

Test build #47593 has finished for PR 10269 at commit bedaead.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class ExecutorClassLoader(\n

[SparkQA]

Test build #47602 has finished for PR 10269 at commit e96556e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[jerryshao]

Thanks @zsxwing for finding out this wrong state transition issue. For now I think assert should be OK.

[zsxwing]

Merging to master & 1.6