Address comments

Author: mccheah

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

@@ -34,15 +34,6 @@ private[spark] object ConfigurationUtils {
     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[_],

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

@@ -45,8 +45,8 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
 
   import ExecutorPodFactoryImpl._
 
-  private val executorExtraClasspath = sparkConf.get(
-    org.apache.spark.internal.config.EXECUTOR_CLASS_PATH)
+  private val executorExtraClasspath =
+    sparkConf.get(org.apache.spark.internal.config.EXECUTOR_CLASS_PATH)
 
   private val executorLabels = ConfigurationUtils.parsePrefixedKeyValuePairs(
     sparkConf,
@@ -59,6 +59,9 @@ private[spark] class ExecutorPodFactoryImpl(sparkConf: SparkConf)
     !executorLabels.contains(SPARK_EXECUTOR_ID_LABEL),
     s"Custom executor labels cannot contain $SPARK_EXECUTOR_ID_LABEL as it is reserved for" +
       s" Spark.")
+  require(
+    !executorLabels.contains(SPARK_ROLE_LABEL),
+    s"Custom executor labels cannot contain $SPARK_ROLE_LABEL as it is reserved for Spark.")
 
   private val executorAnnotations =
     ConfigurationUtils.parsePrefixedKeyValuePairs(

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

@@ -87,12 +87,12 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private val initialExecutors = SchedulerBackendUtils.getInitialTargetExecutorNumber(conf)
 
   private val podAllocationInterval = conf.get(KUBERNETES_ALLOCATION_BATCH_DELAY)
-  require(podAllocationInterval > 0, s"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, s"Allocation batch size " +
+  require(podAllocationSize > 0, "Allocation batch size " +
     s"${KUBERNETES_ALLOCATION_BATCH_SIZE} " +
     s"is ${podAllocationSize}, should be a positive integer")
 

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

@@ -133,8 +133,6 @@ object YarnSparkHadoopUtil {
 
   val ANY_HOST = "*"
 
-  val DEFAULT_NUMBER_EXECUTORS = 2
-
   // All RM requests are issued with same priority : we do not (yet) have any distinction between
   // request types (like map/reduce in hadoop for example)
   val RM_REQUEST_PRIORITY = Priority.newInstance(1)