Addressed more comments

Author: liyinan926

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

@@ -20,7 +20,7 @@ import java.io.File
 
 import com.google.common.base.Charsets
 import com.google.common.io.Files
-import io.fabric8.kubernetes.client.{Config, ConfigBuilder, DefaultKubernetesClient, KubernetesClient}
+import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient, KubernetesClient}
 import io.fabric8.kubernetes.client.utils.HttpClientUtils
 import okhttp3.Dispatcher
 
@@ -88,16 +88,15 @@ private[spark] object SparkKubernetesClientFactory {
     new DefaultKubernetesClient(httpClientWithCustomDispatcher, config)
   }
 
-  private implicit class OptionConfigurableConfigBuilder(configBuilder: ConfigBuilder) {
+  private implicit class OptionConfigurableConfigBuilder(val configBuilder: ConfigBuilder)
+    extends AnyVal {
 
     def withOption[T]
-        (option: Option[T])
-        (configurator: ((T, ConfigBuilder) => ConfigBuilder)): OptionConfigurableConfigBuilder = {
-      new OptionConfigurableConfigBuilder(option.map { opt =>
+    (option: Option[T])
+    (configurator: ((T, ConfigBuilder) => ConfigBuilder)): ConfigBuilder = {
+      option.map { opt =>
         configurator(opt, configBuilder)
-      }.getOrElse(configBuilder))
+      }.getOrElse(configBuilder)
     }
-
-    def build(): Config = configBuilder.build()
   }
 }

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

@@ -16,34 +16,35 @@
  */
 package org.apache.spark.deploy.k8s
 
-package object constants {
+private[spark] object constants {
+
   // Labels
-  private[spark] val SPARK_APP_ID_LABEL = "spark-app-selector"
-  private[spark] val SPARK_EXECUTOR_ID_LABEL = "spark-exec-id"
-  private[spark] val SPARK_ROLE_LABEL = "spark-role"
-  private[spark] val SPARK_POD_DRIVER_ROLE = "driver"
-  private[spark] val SPARK_POD_EXECUTOR_ROLE = "executor"
+  val SPARK_APP_ID_LABEL = "spark-app-selector"
+  val SPARK_EXECUTOR_ID_LABEL = "spark-exec-id"
+  val SPARK_ROLE_LABEL = "spark-role"
+  val SPARK_POD_DRIVER_ROLE = "driver"
+  val SPARK_POD_EXECUTOR_ROLE = "executor"
 
   // Default and fixed ports
-  private[spark] val DEFAULT_DRIVER_PORT = 7078
-  private[spark] val DEFAULT_BLOCKMANAGER_PORT = 7079
-  private[spark] val BLOCK_MANAGER_PORT_NAME = "blockmanager"
-  private[spark] val EXECUTOR_PORT_NAME = "executor"
+  val DEFAULT_DRIVER_PORT = 7078
+  val DEFAULT_BLOCKMANAGER_PORT = 7079
+  val BLOCK_MANAGER_PORT_NAME = "blockmanager"
+  val EXECUTOR_PORT_NAME = "executor"
 
   // Environment Variables
-  private[spark] val ENV_EXECUTOR_PORT = "SPARK_EXECUTOR_PORT"
-  private[spark] val ENV_DRIVER_URL = "SPARK_DRIVER_URL"
-  private[spark] val ENV_EXECUTOR_CORES = "SPARK_EXECUTOR_CORES"
-  private[spark] val ENV_EXECUTOR_MEMORY = "SPARK_EXECUTOR_MEMORY"
-  private[spark] val ENV_APPLICATION_ID = "SPARK_APPLICATION_ID"
-  private[spark] val ENV_EXECUTOR_ID = "SPARK_EXECUTOR_ID"
-  private[spark] val ENV_EXECUTOR_POD_IP = "SPARK_EXECUTOR_POD_IP"
-  private[spark] val ENV_EXECUTOR_EXTRA_CLASSPATH = "SPARK_EXECUTOR_EXTRA_CLASSPATH"
-  private[spark] val ENV_MOUNTED_CLASSPATH = "SPARK_MOUNTED_CLASSPATH"
-  private[spark] val ENV_JAVA_OPT_PREFIX = "SPARK_JAVA_OPT_"
+  val ENV_EXECUTOR_PORT = "SPARK_EXECUTOR_PORT"
+  val ENV_DRIVER_URL = "SPARK_DRIVER_URL"
+  val ENV_EXECUTOR_CORES = "SPARK_EXECUTOR_CORES"
+  val ENV_EXECUTOR_MEMORY = "SPARK_EXECUTOR_MEMORY"
+  val ENV_APPLICATION_ID = "SPARK_APPLICATION_ID"
+  val ENV_EXECUTOR_ID = "SPARK_EXECUTOR_ID"
+  val ENV_EXECUTOR_POD_IP = "SPARK_EXECUTOR_POD_IP"
+  val ENV_EXECUTOR_EXTRA_CLASSPATH = "SPARK_EXECUTOR_EXTRA_CLASSPATH"
+  val ENV_MOUNTED_CLASSPATH = "SPARK_MOUNTED_CLASSPATH"
+  val ENV_JAVA_OPT_PREFIX = "SPARK_JAVA_OPT_"
 
   // Miscellaneous
-  private[spark] val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
-  private[spark] val MEMORY_OVERHEAD_FACTOR = 0.10
-  private[spark] val MEMORY_OVERHEAD_MIN_MIB = 384L
+  val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
+  val MEMORY_OVERHEAD_FACTOR = 0.10
+  val MEMORY_OVERHEAD_MIN_MIB = 384L
 }

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

@@ -61,7 +61,7 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
   require(
     !executorLabels.contains(SPARK_EXECUTOR_ID_LABEL),
     s"Custom executor labels cannot contain $SPARK_EXECUTOR_ID_LABEL as it is reserved for" +
-      s" Spark.")
+      " Spark.")
   require(
     !executorLabels.contains(SPARK_ROLE_LABEL),
     s"Custom executor labels cannot contain $SPARK_ROLE_LABEL as it is reserved for Spark.")
@@ -172,7 +172,7 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
       }
 
     val executorContainer = new ContainerBuilder()
-      .withName(s"executor")
+      .withName("executor")
       .withImage(executorDockerImage)
       .withImagePullPolicy(dockerImagePullPolicy)
       .withNewResources()

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

@@ -330,7 +330,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
           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.")
+              " have failed to start in the first place and never registered with the driver.")
           }
           disconnectedPodsByExecutorIdPendingRemoval.put(executorId, pod)
 
@@ -354,7 +354,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
     private def getExecutorExitStatus(containerStatus: ContainerStatus): Int = {
       Option(containerStatus.getState).map { containerState =>
-        Option(containerState.getTerminated).map {containerStateTerminated =>
+        Option(containerState.getTerminated).map { containerStateTerminated =>
           containerStateTerminated.getExitCode.intValue()
         }.getOrElse(UNKNOWN_EXIT_CODE)
       }.getOrElse(UNKNOWN_EXIT_CODE)
@@ -403,8 +403,8 @@ private[spark] class KubernetesClusterSchedulerBackend(
   }
 
   private class KubernetesDriverEndpoint(
-    rpcEnv: RpcEnv,
-    sparkProperties: Seq[(String, String)])
+      rpcEnv: RpcEnv,
+      sparkProperties: Seq[(String, String)])
     extends DriverEndpoint(rpcEnv, sparkProperties) {
 
     override def onDisconnected(rpcAddress: RpcAddress): Unit = {
@@ -427,4 +427,3 @@ private object KubernetesClusterSchedulerBackend {
   // and assume the executor failed for good, and attribute it to a framework fault.
   val MAX_EXECUTOR_LOST_REASON_CHECKS = 10
 }
-

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala

@@ -39,8 +39,7 @@ import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{Register
 import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
 import org.apache.spark.util.ThreadUtils
 
-class KubernetesClusterSchedulerBackendSuite
-    extends SparkFunSuite with BeforeAndAfter {
+class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAndAfter {
 
   private val APP_ID = "test-spark-app"
   private val DRIVER_POD_NAME = "spark-driver-pod"
@@ -49,7 +48,7 @@ class KubernetesClusterSchedulerBackendSuite
   private val SPARK_DRIVER_PORT = 7077
   private val POD_ALLOCATION_INTERVAL = 60L
   private val DRIVER_URL = RpcEndpointAddress(
-      SPARK_DRIVER_HOST, SPARK_DRIVER_PORT, CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
+    SPARK_DRIVER_HOST, SPARK_DRIVER_PORT, CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
   private val FIRST_EXECUTOR_POD = new PodBuilder()
     .withNewMetadata()
       .withName("pod1")
@@ -75,9 +74,9 @@ class KubernetesClusterSchedulerBackendSuite
 
   private type PODS = MixedOperation[Pod, PodList, DoneablePod, PodResource[Pod, DoneablePod]]
   private type LABELED_PODS = FilterWatchListDeletable[
-      Pod, PodList, java.lang.Boolean, Watch, Watcher[Pod]]
+    Pod, PodList, java.lang.Boolean, Watch, Watcher[Pod]]
   private type IN_NAMESPACE_PODS = NonNamespaceOperation[
-      Pod, PodList, DoneablePod, PodResource[Pod, DoneablePod]]
+    Pod, PodList, DoneablePod, PodResource[Pod, DoneablePod]]
 
   @Mock
   private var sparkContext: SparkContext = _

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -22,15 +22,16 @@ import java.util.concurrent._
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.regex.Pattern
 
-import org.apache.hadoop.yarn.api.records._
-import org.apache.hadoop.yarn.client.api.AMRMClient
-import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
-import org.apache.hadoop.yarn.conf.YarnConfiguration
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet, Queue}
 import scala.util.control.NonFatal
 
+import org.apache.hadoop.yarn.api.records._
+import org.apache.hadoop.yarn.client.api.AMRMClient
+import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
+import org.apache.hadoop.yarn.conf.YarnConfiguration
+
 import org.apache.spark.{SecurityManager, SparkConf, SparkException}
 import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
 import org.apache.spark.deploy.yarn.config._