Addressed more comments

Author: liyinan926

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

@@ -20,6 +20,16 @@ package org.apache.spark.deploy.k8s
 import org.apache.spark.SparkConf
 
 private[spark] object ConfigurationUtils {
+
+  /**
+   * Extract and parse Spark configuration properties with a given name prefix and
+   * return the result as a Map. Keys must not have more than one value.
+   *
+   * @param sparkConf Spark configuration
+   * @param prefix the given property name prefix
+   * @param configType a descriptive note on the type of entities of interest
+   * @return a Map storing the configuration property keys and values
+   */
   def parsePrefixedKeyValuePairs(
       sparkConf: SparkConf,
       prefix: String,
@@ -34,15 +44,6 @@ private[spark] object ConfigurationUtils {
     fromPrefix.toMap
   }
 
-  def requireSecondIfFirstIsDefined(
-      opt1: Option[_],
-      opt2: Option[_],
-      errMessageWhenSecondIsMissing: String): Unit = {
-    opt1.foreach { _ =>
-      require(opt2.isDefined, errMessageWhenSecondIsMissing)
-    }
-  }
-
   def requireNandDefined(opt1: Option[_], opt2: Option[_], errMessage: String): Unit = {
     opt1.foreach { _ => require(opt2.isEmpty, errMessage) }
   }

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

@@ -51,8 +51,8 @@ private[spark] object SparkKubernetesClientFactory {
     ConfigurationUtils.requireNandDefined(
       oauthTokenFile,
       oauthTokenValue,
-      s"Cannot specify OAuth token through both a file $oauthTokenFileConf and a" +
-        s" value $oauthTokenConf.")
+      s"Cannot specify OAuth token through both a file $oauthTokenFileConf and a " +
+        s"value $oauthTokenConf.")
 
     val caCertFile = sparkConf
       .getOption(s"$kubernetesAuthConfPrefix.$CA_CERT_FILE_CONF_SUFFIX")

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

@@ -44,7 +44,6 @@ package object config extends Logging {
       .stringConf
       .createWithDefault("IfNotPresent")
 
-
   private[spark] val APISERVER_AUTH_DRIVER_CONF_PREFIX =
       "spark.kubernetes.authenticate.driver"
   private[spark] val APISERVER_AUTH_DRIVER_MOUNTED_CONF_PREFIX =
@@ -95,12 +94,14 @@ package object config extends Logging {
     ConfigBuilder("spark.kubernetes.allocation.batch.size")
       .doc("Number of pods to launch at once in each round of executor allocation.")
       .intConf
+      .checkValue(value => value > 0, "Allocation batch size should be a positive integer")
       .createWithDefault(5)
 
   private[spark] val KUBERNETES_ALLOCATION_BATCH_DELAY =
     ConfigBuilder("spark.kubernetes.allocation.batch.delay")
       .doc("Number of seconds to wait between each round of executor allocation.")
       .longConf
+      .checkValue(value => value > 0, s"Allocation batch delay should be a positive integer")
       .createWithDefault(1)
 
   private[spark] val KUBERNETES_EXECUTOR_LIMIT_CORES =

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, SparkException}
+import org.apache.spark.SparkConf
 import org.apache.spark.deploy.k8s.ConfigurationUtils
 import org.apache.spark.deploy.k8s.config._
 import org.apache.spark.deploy.k8s.constants._
@@ -77,11 +77,8 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
   private val executorDockerImage = sparkConf.get(EXECUTOR_DOCKER_IMAGE)
   private val dockerImagePullPolicy = sparkConf.get(DOCKER_IMAGE_PULL_POLICY)
   private val executorPort = sparkConf.getInt("spark.executor.port", DEFAULT_STATIC_PORT)
-  private val blockmanagerPort = sparkConf
+  private val blockManagerPort = sparkConf
     .getInt("spark.blockmanager.port", DEFAULT_BLOCKMANAGER_PORT)
-  private val kubernetesDriverPodName = sparkConf
-    .get(KUBERNETES_DRIVER_POD_NAME)
-    .getOrElse(throw new SparkException("Must specify the driver pod name"))
 
   private val executorPodNamePrefix = sparkConf.get(KUBERNETES_EXECUTOR_POD_NAME_PREFIX)
 
@@ -163,7 +160,7 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
     ) ++ executorExtraJavaOptionsEnv ++ executorExtraClasspathEnv.toSeq
     val requiredPorts = Seq(
       (EXECUTOR_PORT_NAME, executorPort),
-      (BLOCK_MANAGER_PORT_NAME, blockmanagerPort))
+      (BLOCK_MANAGER_PORT_NAME, blockManagerPort))
       .map(port => {
         new ContainerPortBuilder()
           .withName(port._1)

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

@@ -36,8 +36,10 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     new TaskSchedulerImpl(sc)
   }
 
-  override def createSchedulerBackend(sc: SparkContext, masterURL: String, scheduler: TaskScheduler)
-      : SchedulerBackend = {
+  override def createSchedulerBackend(
+      sc: SparkContext,
+      masterURL: String,
+      scheduler: TaskScheduler): SchedulerBackend = {
     val sparkConf = sc.getConf
 
     val kubernetesClient = SparkKubernetesClientFactory.createKubernetesClient(

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

@@ -87,14 +87,8 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private val initialExecutors = SchedulerBackendUtils.getInitialTargetExecutorNumber(conf)
 
   private val podAllocationInterval = conf.get(KUBERNETES_ALLOCATION_BATCH_DELAY)
-  require(podAllocationInterval > 0, "Allocation batch delay " +
-    s"${KUBERNETES_ALLOCATION_BATCH_DELAY} " +
-    s"is ${podAllocationInterval}, should be a positive integer")
 
   private val podAllocationSize = conf.get(KUBERNETES_ALLOCATION_BATCH_SIZE)
-  require(podAllocationSize > 0, "Allocation batch size " +
-    s"${KUBERNETES_ALLOCATION_BATCH_SIZE} " +
-    s"is ${podAllocationSize}, should be a positive integer")
 
   private val allocatorRunnable = new Runnable {
 
@@ -304,39 +298,40 @@ private[spark] class KubernetesClusterSchedulerBackend(
     private val DEFAULT_CONTAINER_FAILURE_EXIT_STATUS = -1
 
     override def eventReceived(action: Action, pod: Pod): Unit = {
-      if (action == Action.MODIFIED && pod.getStatus.getPhase == "Running"
-          && pod.getMetadata.getDeletionTimestamp == null) {
-        val podIP = pod.getStatus.getPodIP
-        val clusterNodeName = pod.getSpec.getNodeName
-        logInfo(s"Executor pod $pod ready, launched at $clusterNodeName as IP $podIP.")
-        executorPodsByIPs.put(podIP, pod)
-      } else if (action == Action.DELETED || action == Action.ERROR) {
-        val executorId = pod.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)
-        require(executorId != null, "Unexpected pod metadata; expected all executor pods" +
-          s" to have label $SPARK_EXECUTOR_ID_LABEL.")
-        val podName = pod.getMetadata.getName
-        val podIP = pod.getStatus.getPodIP
-        logDebug(s"Executor pod $podName at IP $podIP was at $action.")
-        if (podIP != null) {
-          executorPodsByIPs.remove(podIP)
-        }
-        val executorExitReason = if (action == Action.ERROR) {
-          logWarning(s"Received pod $podName exited event. Reason: " + pod.getStatus.getReason)
-          executorExitReasonOnError(pod)
-        } else if (action == Action.DELETED) {
-          logWarning(s"Received delete pod $podName event. Reason: " + pod.getStatus.getReason)
-          executorExitReasonOnDelete(pod)
-        } else {
-          throw new IllegalStateException(
-            s"Unknown action that should only be DELETED or ERROR: $action")
-        }
-        podsWithKnownExitReasons.put(pod.getMetadata.getName, executorExitReason)
-        if (!disconnectedPodsByExecutorIdPendingRemoval.containsKey(executorId)) {
-          log.warn(s"Executor with id $executorId was not marked as disconnected, but the" +
-            s" watch received an event of type $action for this executor. The executor may" +
-            s" have failed to start in the first place and never registered with the driver.")
-        }
-        disconnectedPodsByExecutorIdPendingRemoval.put(executorId, pod)
+      action match {
+        case Action.MODIFIED if (pod.getStatus.getPhase == "Running"
+          && pod.getMetadata.getDeletionTimestamp == null) =>
+          val podIP = pod.getStatus.getPodIP
+          val clusterNodeName = pod.getSpec.getNodeName
+          logInfo(s"Executor pod $pod ready, launched at $clusterNodeName as IP $podIP.")
+          executorPodsByIPs.put(podIP, pod)
+        case Action.DELETED | Action.ERROR =>
+          val executorId = getExecutorId(pod)
+          val podName = pod.getMetadata.getName
+          val podIP = pod.getStatus.getPodIP
+          logDebug(s"Executor pod $podName at IP $podIP was at $action.")
+          if (podIP != null) {
+            executorPodsByIPs.remove(podIP)
+          }
+
+          val executorExitReason = if (action == Action.ERROR) {
+            logWarning(s"Received pod $podName exited event. Reason: " + pod.getStatus.getReason)
+            executorExitReasonOnError(pod)
+          } else if (action == Action.DELETED) {
+            logWarning(s"Received delete pod $podName event. Reason: " + pod.getStatus.getReason)
+            executorExitReasonOnDelete(pod)
+          } else {
+            throw new IllegalStateException(
+              s"Unknown action that should only be DELETED or ERROR: $action")
+          }
+          podsWithKnownExitReasons.put(pod.getMetadata.getName, executorExitReason)
+
+          if (!disconnectedPodsByExecutorIdPendingRemoval.containsKey(executorId)) {
+            log.warn(s"Executor with id $executorId was not marked as disconnected, but the" +
+              s" watch received an event of type $action for this executor. The executor may" +
+              s" have failed to start in the first place and never registered with the driver.")
+          }
+          disconnectedPodsByExecutorIdPendingRemoval.put(executorId, pod)
       }
     }
 
@@ -391,6 +386,13 @@ private[spark] class KubernetesClusterSchedulerBackend(
       ExecutorExited(
         getExecutorExitStatus(pod), exitCausedByApp = false, exitMessage)
     }
+
+    def getExecutorId(pod: Pod): String = {
+      val executorId = pod.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)
+      require(executorId != null, "Unexpected pod metadata; expected all executor pods " +
+        s"to have label $SPARK_EXECUTOR_ID_LABEL.")
+      executorId
+    }
   }
 
   override def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {