Stop LoggingPodStatusWatcher when unable to start a watch

[ash211]

Saw this exception when the deploy client was attempting to start a watch and unable to (because the apiserver was unroutable)

When seeing exceptions like this when starting the watch, we should be sure the watch gets closed.

Possibly this is an upstream concern, where all watches that are attempted to start should also receive an onClose() callback with the setup exception.

07:10:59.256 [pool-1-thread-1] INFO  o.a.s.d.k.LoggingPodStatusWatcher - Application status for spark-app (phase: unknown)
Exception in thread “main” io.fabric8.kubernetes.client.KubernetesClientException: An error has occurred.
  at io.fabric8.kubernetes.client.KubernetesClientException.launderThrowable(KubernetesClientException.java:61)
  at io.fabric8.kubernetes.client.KubernetesClientException.launderThrowable(KubernetesClientException.java:52)
  at io.fabric8.kubernetes.client.dsl.base.BaseOperation.list(BaseOperation.java:543)
  at io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager.<init>(WatchConnectionManager.java:84)
  at io.fabric8.kubernetes.client.dsl.base.BaseOperation.watch(BaseOperation.java:650)
  at io.fabric8.kubernetes.client.dsl.base.BaseOperation.watch(BaseOperation.java:638)
  at io.fabric8.kubernetes.client.dsl.base.BaseOperation.watch(BaseOperation.java:68)
  at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$9$$anonfun$apply$7.apply(Client.scala:151)
  at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$9$$anonfun$apply$7.apply(Client.scala:151)
  at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2535)
  at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$9.apply(Client.scala:151)
  at org.apache.spark.deploy.kubernetes.Client$$anonfun$run$9.apply(Client.scala:140)
  at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2536)
  at org.apache.spark.deploy.kubernetes.Client.run(Client.scala:140)
  at org.apache.spark.deploy.kubernetes.Client$.main(Client.scala:727)
  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:498)
  at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:758)
  at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:179)
  at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:204)
  at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:118)
  at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.net.SocketTimeoutException: connect timed out
  at java.net.PlainSocketImpl.socketConnect(Native Method)
  at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
  at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
  at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
  at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
  at java.net.Socket.connect(Socket.java:589)
  at okhttp3.internal.platform.Platform.connectSocket(Platform.java:124)
  at okhttp3.internal.connection.RealConnection.connectSocket(RealConnection.java:220)
  at okhttp3.internal.connection.RealConnection.connect(RealConnection.java:146)
  at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:186)
  at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:121)
  at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:100)
  at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67)
  at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67)
  at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92)
  at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:120)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67)
  at io.fabric8.kubernetes.client.utils.HttpClientUtils$2.intercept(HttpClientUtils.java:93)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92)
  at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67)
  at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:179)
  at okhttp3.RealCall.execute(RealCall.java:63)
  at io.fabric8.kubernetes.client.dsl.base.OperationSupport.handleResponse(OperationSupport.java:237)
  at io.fabric8.kubernetes.client.dsl.base.OperationSupport.handleResponse(OperationSupport.java:232)
  at io.fabric8.kubernetes.client.dsl.base.BaseOperation.list(BaseOperation.java:541)
  ... 22 more

[lins05]

Also encountered this earlier today when i tried to launch the example app against a local k8s cluster powered by vagrant but forgot to run "vagrant up" before that.