Add Support for some kubernetes client settings

Add Support for `spark.kubernetes.client.watch.reconnectInterval`(which is relevant to KUBERNETES_WATCH_RECONNECT_INTERVAL_SYSTEM_PROPERTY in io.fabric8.kubernetes.client.Config.)
Add Support for `spark.kubernetes.client.watch.reconnectLimit`(which is relevant to KUBERNETES_WATCH_RECONNECT_LIMIT_SYSTEM_PROPERTY in io.fabric8.kubernetes.client.Config.)
Add Support for `spark.kubernetes.client.connection.timeout`(which is relevant to KUBERNETES_CONNECTION_TIMEOUT_SYSTEM_PROPERTY in io.fabric8.kubernetes.client.Config.)
Add Support for `spark.kubernetes.client.request.timeout`(which is relevant to KUBERNETES_REQUEST_TIMEOUT_SYSTEM_PROPERTY in io.fabric8.kubernetes.client.Config.)

Signed-off-by: duyanghao <1294057873@qq.com>

Author: duyanghao

Diff for: docs/running-on-kubernetes.md

@@ -798,6 +798,34 @@ from the other deployment modes. See the [configuration page](configuration.html
     the Driver process. The user can specify multiple of these to set multiple environment variables.
   </td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.client.watch.reconnectIntervalInMs</code></td>
+  <td><code>1s</code></td>
+  <td>
+    Connection retry interval for kubernetes client requests.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.client.watch.reconnectLimit</code></td>
+  <td><code>-1</code></td>
+  <td>
+    Limit of times connections can be attempted for kubernetes client requests.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.client.connection.timeoutInMs</code></td>
+  <td><code>10s</code></td>
+  <td>
+    Connection timeout for kubernetes client requests.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.client.request.timeoutInMs</code></td>
+  <td><code>10s</code></td>
+  <td>
+    Request timeout for kubernetes client requests.
+  </td>
+</tr>
 </table>
 
 

Diff for: resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/SparkKubernetesClientFactory.scala

@@ -81,6 +81,18 @@ private[spark] object SparkKubernetesClientFactory {
         }.withOption(namespace) {
           (ns, configBuilder) => configBuilder.withNamespace(ns)
         }.build()
+    val connTimeoutPropertiesMs = sparkConf.get(
+        KUBERNETES_CLIENT_CONNECTION_TIMEOUT_SYSTEM_PROPERTY)
+    config.setConnectionTimeout(connTimeoutPropertiesMs.toInt)
+    val reqTimeoutPropertiesMs = sparkConf.get(
+        KUBERNETES_CLIENT_REQUEST_TIMEOUT_SYSTEM_PROPERTY)
+    config.setRequestTimeout(reqTimeoutPropertiesMs.toInt)
+    val reconnectIntervalPropertiesMs = sparkConf.get(
+        KUBERNETES_CLIENT_WATCH_RECONNECT_INTERVAL_SYSTEM_PROPERTY)
+    config.setWatchReconnectInterval(reconnectIntervalPropertiesMs.toInt)
+    val reconnectLimitProperties = sparkConf.get(
+        KUBERNETES_CLIENT_WATCH_RECONNECT_LIMIT_SYSTEM_PROPERTY)
+    config.setWatchReconnectLimit(reconnectLimitProperties)
     val baseHttpClient = HttpClientUtils.createHttpClient(config)
     val httpClientWithCustomDispatcher = baseHttpClient.newBuilder()
       .dispatcher(dispatcher)

Diff for: resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala

@@ -545,4 +545,28 @@ package object config extends Logging {
       resolvedURL
     }
   }
+
+  private[spark] val KUBERNETES_CLIENT_WATCH_RECONNECT_INTERVAL_SYSTEM_PROPERTY =
+    ConfigBuilder("spark.kubernetes.client.watch.reconnectIntervalInMs")
+      .doc("Connection retry interval for kubernetes client requests.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("1s")
+
+  private[spark] val KUBERNETES_CLIENT_WATCH_RECONNECT_LIMIT_SYSTEM_PROPERTY =
+    ConfigBuilder("spark.kubernetes.client.watch.reconnectLimit")
+      .doc("Limit of times connections can be attempted for kubernetes client requests.")
+      .intConf
+      .createWithDefault(-1)
+
+  private[spark] val KUBERNETES_CLIENT_CONNECTION_TIMEOUT_SYSTEM_PROPERTY =
+    ConfigBuilder("spark.kubernetes.client.connection.timeoutInMs")
+      .doc("Connection timeout for kubernetes client requests.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("10s")
+
+  private[spark] val KUBERNETES_CLIENT_REQUEST_TIMEOUT_SYSTEM_PROPERTY =
+    ConfigBuilder("spark.kubernetes.client.request.timeoutInMs")
+      .doc("Request timeout for kubernetes client requests.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("10s")
 }