Launcher doesn't stop when SparkException thrown

[ash211]

As reported by @kimoonkim at #120 (comment):

Found another case Launcher did not stop. My k8s cluster uses an overlay network, which was broken because the overlay daemons crashed. This prevented the launcher from uploading local jars. The good news is that our Client code correctly recognized this and shut down the pod and service:

Exception in thread "main" org.apache.spark.SparkException: Failed to submit the application to the driver pod.
	at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$7$$anonfun$apply$6.apply(Client.scala:169)
	at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$7$$anonfun$apply$6.apply(Client.scala:131)
	at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2530)
	at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$7.apply(Client.scala:131)
	at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$7.apply(Client.scala:109)
	at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2530)
	at org.apache.spark.deploy.kubernetes.Client.run(Client.scala:109)
	at org.apache.spark.deploy.kubernetes.Client$.main(Client.scala:812)
	at org.apache.spark.deploy.kubernetes.Client.main(Client.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:497)
	at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:747)
	at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:178)
	at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:203)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:117)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

However, LoggingPodStatusWatcher was not informed of this and kept going:

2017-02-23 13:00:06 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:07 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:08 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:09 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:10 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:11 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:12 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
2017-02-23 13:00:13 INFO  LoggingPodStatusWatcher:54 - Application status for spark-pi-1487883497166 (phase: Running)
...

[kimoonkim]

@ash211 Thanks for filing this issue.

I have a local workaround that does not depend on the k8s client fix. It just shuts down LoggingPodStatusWatcher explicitly as well. Please let me know if we want this as a PR.

[ash211]

@kimoonkim if you have a PR that works around this please send the PR -- we should make Spark resilient to this issue even if there are bugs upstream in the client.

[kimoonkim]

With #154 by foxish, my upcoming PR seems less important. But I'll send it anyway to see what we think. The remaining benefit is that it will work even when watcher couldn't detect the DELETED pod state due to upstream client bugs.

[kimoonkim]

In today's weekly SIG meeting, we were talking about potentially reviving #160 to work around fabric8io/kubernetes-client#674. It occurred to me there is another approach. We can make the watcher thread a daemon thread. JVM ensures to shut down all daemon threads when non-daemon threads (typically, just the main thread) completes. This may achieve the same goal of #160 in a less intrusive way. And it can be deemed the right thing to do anyway.

@ash211 @mccheah Let me know what you think. Turning it into a daemon thread is a small change. One just needs to write a simple thread factory that sets the daemon thread attribute on a new thread.