Address a bunch of style and other comments

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

@@ -17,10 +17,9 @@
 
 package org.apache.spark.deploy.k8s
 
-import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.internal.Logging
+import org.apache.spark.SparkConf
 
-private[spark] object ConfigurationUtils extends Logging {
+private[spark] object ConfigurationUtils {
   def parsePrefixedKeyValuePairs(
       sparkConf: SparkConf,
       prefix: String,
@@ -34,4 +33,24 @@ private[spark] object ConfigurationUtils extends Logging {
     }
     fromPrefix.toMap
   }
+
+  def requireBothOrNeitherDefined(
+    opt1: Option[_],
+    opt2: Option[_],
+    errMessageWhenFirstIsMissing: String,
+    errMessageWhenSecondIsMissing: String): Unit = {
+    requireSecondIfFirstIsDefined(opt1, opt2, errMessageWhenSecondIsMissing)
+    requireSecondIfFirstIsDefined(opt2, opt1, errMessageWhenFirstIsMissing)
+  }
+
+  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

@@ -48,7 +48,7 @@ private[spark] object SparkKubernetesClientFactory {
       .map(new File(_))
       .orElse(maybeServiceAccountToken)
     val oauthTokenValue = sparkConf.getOption(oauthTokenConf)
-    OptionRequirements.requireNandDefined(
+    ConfigurationUtils.requireNandDefined(
         oauthTokenFile,
         oauthTokenValue,
         s"Cannot specify OAuth token through both a file $oauthTokenFileConf and a" +

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

@@ -31,12 +31,6 @@ package object config extends Logging {
       .stringConf
       .createWithDefault("default")
 
-  private[spark] val DRIVER_DOCKER_IMAGE =
-    ConfigBuilder("spark.kubernetes.driver.docker.image")
-      .doc("Docker image to use for the driver. Specify this using the standard Docker tag format.")
-      .stringConf
-      .createWithDefault(s"spark-driver:$sparkVersion")
-
   private[spark] val EXECUTOR_DOCKER_IMAGE =
     ConfigBuilder("spark.kubernetes.executor.docker.image")
       .doc("Docker image to use for the executors. Specify this using the standard Docker tag" +
@@ -81,22 +75,8 @@ package object config extends Logging {
       .bytesConf(ByteUnit.MiB)
       .createOptional
 
-  private[spark] val KUBERNETES_DRIVER_MEMORY_OVERHEAD =
-    ConfigBuilder("spark.kubernetes.driver.memoryOverhead")
-      .doc("The amount of off-heap memory (in megabytes) to be allocated for the driver and the" +
-        " driver submission server. This is memory that accounts for things like VM overheads," +
-        " interned strings, other native overheads, etc. This tends to grow with the driver's" +
-        " memory size (typically 6-10%).")
-      .bytesConf(ByteUnit.MiB)
-      .createOptional
-
-  private[spark] val KUBERNETES_DRIVER_LABEL_PREFIX = "spark.kubernetes.driver.label."
-  private[spark] val KUBERNETES_DRIVER_ANNOTATION_PREFIX = "spark.kubernetes.driver.annotation."
   private[spark] val KUBERNETES_EXECUTOR_LABEL_PREFIX = "spark.kubernetes.executor.label."
   private[spark] val KUBERNETES_EXECUTOR_ANNOTATION_PREFIX = "spark.kubernetes.executor.annotation."
-  private[spark] val KUBERNETES_DRIVER_ENV_KEY = "spark.kubernetes.driverEnv."
-  private[spark] val KUBERNETES_DRIVER_SECRETS_PREFIX = "spark.kubernetes.driver.secrets."
-  private[spark] val KUBERNETES_EXECUTOR_SECRETS_PREFIX = "spark.kubernetes.executor.secrets."
 
   private[spark] val KUBERNETES_DRIVER_POD_NAME =
     ConfigBuilder("spark.kubernetes.driver.pod.name")

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

@@ -24,29 +24,10 @@ package object constants {
   private[spark] val SPARK_POD_DRIVER_ROLE = "driver"
   private[spark] val SPARK_POD_EXECUTOR_ROLE = "executor"
 
-  // Credentials secrets
-  private[spark] val DRIVER_CREDENTIALS_SECRETS_BASE_DIR =
-    "/mnt/secrets/spark-kubernetes-credentials"
-  private[spark] val DRIVER_CREDENTIALS_CA_CERT_SECRET_NAME = "ca-cert"
-  private[spark] val DRIVER_CREDENTIALS_CA_CERT_PATH =
-    s"$DRIVER_CREDENTIALS_SECRETS_BASE_DIR/$DRIVER_CREDENTIALS_CA_CERT_SECRET_NAME"
-  private[spark] val DRIVER_CREDENTIALS_CLIENT_KEY_SECRET_NAME = "client-key"
-  private[spark] val DRIVER_CREDENTIALS_CLIENT_KEY_PATH =
-    s"$DRIVER_CREDENTIALS_SECRETS_BASE_DIR/$DRIVER_CREDENTIALS_CLIENT_KEY_SECRET_NAME"
-  private[spark] val DRIVER_CREDENTIALS_CLIENT_CERT_SECRET_NAME = "client-cert"
-  private[spark] val DRIVER_CREDENTIALS_CLIENT_CERT_PATH =
-    s"$DRIVER_CREDENTIALS_SECRETS_BASE_DIR/$DRIVER_CREDENTIALS_CLIENT_CERT_SECRET_NAME"
-  private[spark] val DRIVER_CREDENTIALS_OAUTH_TOKEN_SECRET_NAME = "oauth-token"
-  private[spark] val DRIVER_CREDENTIALS_OAUTH_TOKEN_PATH =
-    s"$DRIVER_CREDENTIALS_SECRETS_BASE_DIR/$DRIVER_CREDENTIALS_OAUTH_TOKEN_SECRET_NAME"
-  private[spark] val DRIVER_CREDENTIALS_SECRET_VOLUME_NAME = "kubernetes-credentials"
-
   // Default and fixed ports
   private[spark] val DEFAULT_DRIVER_PORT = 7078
   private[spark] val DEFAULT_BLOCKMANAGER_PORT = 7079
-  private[spark] val DEFAULT_UI_PORT = 4040
   private[spark] val BLOCK_MANAGER_PORT_NAME = "blockmanager"
-  private[spark] val DRIVER_PORT_NAME = "driver-rpc-port"
   private[spark] val EXECUTOR_PORT_NAME = "executor"
 
   // Environment Variables
@@ -57,20 +38,11 @@ package object constants {
   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_DRIVER_MEMORY = "SPARK_DRIVER_MEMORY"
-  private[spark] val ENV_SUBMIT_EXTRA_CLASSPATH = "SPARK_SUBMIT_EXTRA_CLASSPATH"
   private[spark] val ENV_EXECUTOR_EXTRA_CLASSPATH = "SPARK_EXECUTOR_EXTRA_CLASSPATH"
   private[spark] val ENV_MOUNTED_CLASSPATH = "SPARK_MOUNTED_CLASSPATH"
-  private[spark] val ENV_DRIVER_MAIN_CLASS = "SPARK_DRIVER_CLASS"
-  private[spark] val ENV_DRIVER_ARGS = "SPARK_DRIVER_ARGS"
-  private[spark] val ENV_DRIVER_JAVA_OPTS = "SPARK_DRIVER_JAVA_OPTS"
-  private[spark] val ENV_MOUNTED_FILES_DIR = "SPARK_MOUNTED_FILES_DIR"
-  private[spark] val ENV_PYSPARK_FILES = "PYSPARK_FILES"
-  private[spark] val ENV_PYSPARK_PRIMARY = "PYSPARK_PRIMARY"
   private[spark] val ENV_JAVA_OPT_PREFIX = "SPARK_JAVA_OPT_"
 
   // Miscellaneous
-  private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   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

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

@@ -18,8 +18,7 @@ package org.apache.spark.scheduler.cluster.k8s
 
 import scala.collection.JavaConverters._
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, ContainerPortBuilder, EnvVar, EnvVarBuilder, EnvVarSourceBuilder, Pod, PodBuilder, QuantityBuilder}
-import org.apache.commons.io.FilenameUtils
+import io.fabric8.kubernetes.api.model._
 
 import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.ConfigurationUtils
@@ -48,7 +47,7 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
       org.apache.spark.internal.config.EXECUTOR_CLASS_PATH)
   private val executorJarsDownloadDir = sparkConf.get(INIT_CONTAINER_JARS_DOWNLOAD_LOCATION)
 
-  private val executorLabels = ConfigurationUtils.parsePrefixedKeyValuePairs (
+  private val executorLabels = ConfigurationUtils.parsePrefixedKeyValuePairs(
       sparkConf,
       KUBERNETES_EXECUTOR_LABEL_PREFIX,
       "executor label")

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

@@ -21,13 +21,12 @@ import java.io.File
 import io.fabric8.kubernetes.client.Config
 
 import org.apache.spark.SparkContext
-import org.apache.spark.deploy.k8s.{ConfigurationUtils, SparkKubernetesClientFactory}
+import org.apache.spark.deploy.k8s.SparkKubernetesClientFactory
 import org.apache.spark.deploy.k8s.config._
 import org.apache.spark.deploy.k8s.constants._
 import org.apache.spark.internal.Logging
-import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.scheduler.{ExternalClusterManager, SchedulerBackend, TaskScheduler, TaskSchedulerImpl}
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.ThreadUtils
 
 private[spark] class KubernetesClusterManager extends ExternalClusterManager with Logging {
 

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

@@ -21,8 +21,8 @@ import java.net.InetAddress
 import java.util.concurrent.{ConcurrentHashMap, ExecutorService, ScheduledExecutorService, TimeUnit}
 import java.util.concurrent.atomic.{AtomicInteger, AtomicLong, AtomicReference}
 
-import scala.collection.{concurrent, mutable}
 import scala.collection.JavaConverters._
+import scala.collection.mutable
 import scala.concurrent.{ExecutionContext, Future}
 
 import io.fabric8.kubernetes.api.model._

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

@@ -18,20 +18,14 @@ package org.apache.spark.scheduler.cluster.k8s
 
 import scala.collection.JavaConverters._
 
-import io.fabric8.kubernetes.api.model.{Pod, VolumeBuilder, VolumeMountBuilder, _}
+import io.fabric8.kubernetes.api.model.{Pod, _}
 import io.fabric8.kubernetes.client.KubernetesClient
-import org.apache.commons.io.FilenameUtils
-import org.mockito.{AdditionalAnswers, MockitoAnnotations}
-import org.mockito.Matchers.{any, eq => mockitoEq}
-import org.mockito.Mockito._
-import org.mockito.invocation.InvocationOnMock
-import org.mockito.stubbing.Answer
+import org.mockito.MockitoAnnotations
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterEach}
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s.config._
 import org.apache.spark.deploy.k8s.constants
-import org.apache.spark.scheduler.cluster.k8s.ExecutorPodFactoryImpl
 
 class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with BeforeAndAfterEach {
   private val driverPodName: String = "driver-pod"

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

@@ -27,7 +27,7 @@ import io.fabric8.kubernetes.client.Watcher.Action
 import io.fabric8.kubernetes.client.dsl.{FilterWatchListDeletable, MixedOperation, NonNamespaceOperation, PodResource}
 import org.mockito.{AdditionalAnswers, ArgumentCaptor, Mock, MockitoAnnotations}
 import org.mockito.Matchers.{any, eq => mockitoEq}
-import org.mockito.Mockito.{mock => _, _}
+import org.mockito.Mockito.{doNothing, never, times, verify, when}
 import org.scalatest.BeforeAndAfter
 import org.scalatest.mock.MockitoSugar._
 
@@ -74,7 +74,7 @@ private[spark] class KubernetesClusterSchedulerBackendSuite
     .build()
 
   private type PODS = MixedOperation[Pod, PodList, DoneablePod, PodResource[Pod, DoneablePod]]
-  private type LABELLED_PODS = FilterWatchListDeletable[
+  private type LABELED_PODS = FilterWatchListDeletable[
       Pod, PodList, java.lang.Boolean, Watch, Watcher[Pod]]
   private type IN_NAMESPACE_PODS = NonNamespaceOperation[
       Pod, PodList, DoneablePod, PodResource[Pod, DoneablePod]]
@@ -104,7 +104,7 @@ private[spark] class KubernetesClusterSchedulerBackendSuite
   private var podOperations: PODS = _
 
   @Mock
-  private var podsWithLabelOperations: LABELLED_PODS = _
+  private var podsWithLabelOperations: LABELED_PODS = _
 
   @Mock
   private var podsInNamespace: IN_NAMESPACE_PODS = _