KY-895, add config KUBERNETES_DELETE_DRIVER (apache#376)

Author: fishcus

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -459,6 +459,14 @@ private[spark] object Config extends Logging {
       .checkValue(delay => delay > 0, "delay must be a positive time value")
       .createWithDefaultString("30s")
 
+  val KUBERNETES_DELETE_DRIVER =
+    ConfigBuilder("spark.kubernetes.driver.deleteOnCompleted")
+      .doc("If set to false then driver pods will not be deleted in case " +
+        "of completion.")
+      .version("3.1.1")
+      .booleanConf
+      .createWithDefault(true)
+
   val KUBERNETES_DRIVER_LABEL_PREFIX = "spark.kubernetes.driver.label."
   val KUBERNETES_DRIVER_ANNOTATION_PREFIX = "spark.kubernetes.driver.annotation."
   val KUBERNETES_DRIVER_SERVICE_ANNOTATION_PREFIX = "spark.kubernetes.driver.service.annotation."

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

@@ -107,7 +107,9 @@ private[k8s] class LoggingPodStatusWatcherImpl(conf: KubernetesDriverConf,
         pod.map { p => s"Container final statuses:\n\n${containersDescription(p)}" }
           .getOrElse("No containers were found in the driver pod."))
       logInfo(s"Application ${conf.appName} with submission ID $sId finished")
-      pod.map { p => kubernetesClient.pods().withName(p.getMetadata.getName).delete() }
+      if (conf.get(KUBERNETES_DELETE_DRIVER)) {
+        pod.map { p => kubernetesClient.pods().withName(p.getMetadata.getName).delete() }
+      }
     }
     podCompleted
   } else {