Addressed one more round of comments

Author: liyinan926

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -668,7 +668,9 @@ private[spark] object SparkConf extends Logging {
     MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM.key -> Seq(
       AlternateConfig("spark.reducer.maxReqSizeShuffleToMem", "2.3")),
     LISTENER_BUS_EVENT_QUEUE_CAPACITY.key -> Seq(
-      AlternateConfig("spark.scheduler.listenerbus.eventqueue.size", "2.3"))
+      AlternateConfig("spark.scheduler.listenerbus.eventqueue.size", "2.3")),
+    "spark.driver.memoryOverhead" -> Seq(
+      AlternateConfig("spark.yarn.driver.memoryOverhead", "2.3"))
   )
 
   /**

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -2751,16 +2751,29 @@ private[spark] object Utils extends Logging {
     require(rawMasterURL.startsWith("k8s://"),
       "Kubernetes master URL must start with k8s://.")
     val masterWithoutK8sPrefix = rawMasterURL.substring("k8s://".length)
-    if (masterWithoutK8sPrefix.startsWith("https://")) {
-      masterWithoutK8sPrefix
-    } else if (masterWithoutK8sPrefix.startsWith("http://")) {
-      logWarning("Kubernetes master URL uses HTTP instead of HTTPS.")
-      masterWithoutK8sPrefix
-    } else {
+
+    // To handle master URLs, e.g., k8s://host:port.
+    if (!masterWithoutK8sPrefix.contains("://")) {
       val resolvedURL = s"https://$masterWithoutK8sPrefix"
       logInfo("No scheme specified for kubernetes master URL, so defaulting to https. Resolved " +
         s"URL is $resolvedURL.")
-      resolvedURL
+      return resolvedURL
+    }
+
+    val masterScheme = new URI(masterWithoutK8sPrefix).getScheme
+    masterScheme.toLowerCase match {
+      case "https" =>
+        masterWithoutK8sPrefix
+      case "http" =>
+        logWarning("Kubernetes master URL uses HTTP instead of HTTPS.")
+        masterWithoutK8sPrefix
+      case null =>
+        val resolvedURL = s"https://$masterWithoutK8sPrefix"
+        logInfo("No scheme specified for kubernetes master URL, so defaulting to https. Resolved " +
+          s"URL is $resolvedURL.")
+        resolvedURL
+      case _ =>
+        throw new IllegalArgumentException("Invalid Kubernetes master scheme: " + masterScheme)
     }
   }
 }

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -408,6 +408,7 @@ class SparkSubmitSuite
     childArgsMap.get("--arg") should be (Some("arg1"))
     mainClass should be (KUBERNETES_CLUSTER_SUBMIT_CLASS)
     classpath should have length (0)
+    conf.get("spark.master") should be ("https://host:port")
     conf.get("spark.executor.memory") should be ("5g")
     conf.get("spark.driver.memory") should be ("4g")
     conf.get("spark.kubernetes.namespace") should be ("spark")

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -1153,9 +1153,19 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties with Logging {
     val k8sMasterURLHttp = Utils.checkAndGetK8sMasterUrl("k8s://http://host:port")
     assert(k8sMasterURLHttp == "http://host:port")
 
+    val k8sMasterURLWithoutScheme = Utils.checkAndGetK8sMasterUrl("k8s://127.0.0.1:8443")
+    assert(k8sMasterURLWithoutScheme == "https://127.0.0.1:8443")
+
+    val k8sMasterURLWithoutScheme2 = Utils.checkAndGetK8sMasterUrl("k8s://127.0.0.1")
+    assert(k8sMasterURLWithoutScheme2 == "https://127.0.0.1")
+
     intercept[IllegalArgumentException] {
       Utils.checkAndGetK8sMasterUrl("k8s:https://host:port")
     }
+
+    intercept[IllegalArgumentException] {
+      Utils.checkAndGetK8sMasterUrl("k8s://foo://host:port")
+    }
   }
 }
 

launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java

@@ -76,9 +76,6 @@ class SparkSubmitOptionParser {
   protected final String PRINCIPAL = "--principal";
   protected final String QUEUE = "--queue";
 
-  // Kubernetes-only options.
-  protected final String KUBERNETES_NAMESPACE = "--kubernetes-namespace";
-
   /**
    * This is the canonical list of spark-submit options. Each entry in the array contains the
    * different aliases for the same option; the first element of each entry is the "official"
@@ -117,8 +114,7 @@ class SparkSubmitOptionParser {
     { QUEUE },
     { REPOSITORIES },
     { STATUS },
-    { TOTAL_EXECUTOR_CORES },
-    { KUBERNETES_NAMESPACE },
+    { TOTAL_EXECUTOR_CORES }
   };
 
   /**

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

@@ -20,6 +20,7 @@ import java.util.{Collections, UUID}
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
+import scala.util.control.NonFatal
 
 import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.client.KubernetesClient
@@ -137,10 +138,10 @@ private[spark] class Client(
       .build()
 
     Utils.tryWithResource(
-        kubernetesClient
-          .pods()
-          .withName(resolvedDriverPod.getMetadata.getName)
-          .watch(loggingPodStatusWatcher)) { _ =>
+      kubernetesClient
+        .pods()
+        .withName(resolvedDriverPod.getMetadata.getName)
+        .watch(loggingPodStatusWatcher)) { _ =>
       val createdDriverPod = kubernetesClient.pods().create(resolvedDriverPod)
       try {
         if (currentDriverSpec.otherKubernetesResources.nonEmpty) {
@@ -149,7 +150,7 @@ private[spark] class Client(
           kubernetesClient.resourceList(otherKubernetesResources: _*).createOrReplace()
         }
       } catch {
-        case e: Throwable =>
+        case NonFatal(e) =>
           kubernetesClient.pods().delete(createdDriverPod)
           throw e
       }