Removed default value for executor Docker image

Author: liyinan926

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

@@ -16,7 +16,6 @@
  */
 package org.apache.spark.deploy.k8s
 
-import org.apache.spark.SPARK_VERSION
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.ConfigBuilder
 import org.apache.spark.network.util.ByteUnit
@@ -36,7 +35,7 @@ private[spark] object Config extends Logging {
       .doc("Docker image to use for the executors. Specify this using the standard Docker tag" +
         " format.")
       .stringConf
-      .createWithDefault(s"spark-executor:$SPARK_VERSION")
+      .createOptional
 
   val DOCKER_IMAGE_PULL_POLICY =
     ConfigBuilder("spark.kubernetes.docker.image.pullPolicy")

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala

@@ -20,7 +20,7 @@ import scala.collection.JavaConverters._
 
 import io.fabric8.kubernetes.api.model._
 
-import org.apache.spark.SparkConf
+import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.ConfigurationUtils
 import org.apache.spark.deploy.k8s.Constants._
@@ -72,7 +72,9 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
       sparkConf,
       KUBERNETES_NODE_SELECTOR_PREFIX)
 
-  private val executorDockerImage = sparkConf.get(EXECUTOR_DOCKER_IMAGE)
+  private val executorDockerImage = sparkConf
+    .get(EXECUTOR_DOCKER_IMAGE)
+    .getOrElse(throw new SparkException("Must specify the executor Docker image"))
   private val dockerImagePullPolicy = sparkConf.get(DOCKER_IMAGE_PULL_POLICY)
   private val blockManagerPort = sparkConf
     .getInt("spark.blockmanager.port", DEFAULT_BLOCKMANAGER_PORT)