Append test and fix typo

docs/running-on-kubernetes.md

@@ -1332,21 +1332,30 @@ See the [configuration page](configuration.html) for information on Spark config
   <td>3.3.0</td>
 </tr>
 <tr>
-  <td><code>spark.kubernetes.executor.scheduler.name<code></td>
+  <td><code>spark.kubernetes.executor.scheduler.name</code></td>
   <td>(none)</td>
   <td>
 	Specify the scheduler name for each executor pod.
   </td>
   <td>3.0.0</td>
 </tr>
 <tr>
-  <td><code>spark.kubernetes.driver.scheduler.name<code></td>
+  <td><code>spark.kubernetes.driver.scheduler.name</code></td>
   <td>(none)</td>
   <td>
     Specify the scheduler name for driver pod.
   </td>
   <td>3.3.0</td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.scheduler.name</code></td>
+  <td>(none)</td>
+  <td>
+    Specify the scheduler name for driver and executor pods. If `spark.kubernetes.driver.scheduler.name` or
+    `spark.kubernetes.executor.scheduler.name` is set, will override this.
+  </td>
+  <td>3.3.0</td>
+</tr>
 <tr>
   <td><code>spark.kubernetes.configMap.maxSize</code></td>
   <td><code>1572864</code></td>

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

@@ -285,7 +285,7 @@ private[spark] object Config extends Logging {
 
   val KUBERNETES_SCHEDULER_NAME =
     ConfigBuilder("spark.kubernetes.scheduler.name")
-      .doc("Specify the scheduler name for driver and executor pods. if " +
+      .doc("Specify the scheduler name for driver and executor pods. If " +
         s"`${KUBERNETES_DRIVER_SCHEDULER_NAME.key}` or " +
         s"`${KUBERNETES_EXECUTOR_SCHEDULER_NAME.key}` is set, will override this.")
       .version("3.3.0")

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/KubernetesConfSuite.scala

@@ -215,13 +215,13 @@ class KubernetesConfSuite extends SparkFunSuite {
     assert(KubernetesTestConf.createDriverConf(sparkConf).schedulerName === Some("sameScheduler"))
     assert(KubernetesTestConf.createExecutorConf(sparkConf).schedulerName === Some("sameScheduler"))
 
-    // Overwrite by driver/executor side scheduler when ""
+    // Override by driver/executor side scheduler when ""
     sparkConf.set(KUBERNETES_DRIVER_SCHEDULER_NAME, "")
     sparkConf.set(KUBERNETES_EXECUTOR_SCHEDULER_NAME, "")
     assert(KubernetesTestConf.createDriverConf(sparkConf).schedulerName === Some(""))
     assert(KubernetesTestConf.createExecutorConf(sparkConf).schedulerName === Some(""))
 
-    // Overwrite by driver/executor side scheduler when set
+    // Override by driver/executor side scheduler when set
     sparkConf.set(KUBERNETES_DRIVER_SCHEDULER_NAME, "driverScheduler")
     sparkConf.set(KUBERNETES_EXECUTOR_SCHEDULER_NAME, "executorScheduler")
     val execConf = KubernetesTestConf.createExecutorConf(sparkConf)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/PodBuilderSuite.scala

@@ -36,6 +36,8 @@ abstract class PodBuilderSuite extends SparkFunSuite {
 
   protected def templateFileConf: ConfigEntry[_]
 
+  protected def roleSpecificSchedulerNameConf: ConfigEntry[_]
+
   protected def userFeatureStepsConf: ConfigEntry[_]
 
   protected def userFeatureStepWithExpectedAnnotation: (String, String)
@@ -55,10 +57,16 @@ abstract class PodBuilderSuite extends SparkFunSuite {
 
   test("SPARK-36059: set custom scheduler") {
     val client = mockKubernetesClient()
-    val sparkConf = baseConf.clone().set(templateFileConf.key, "template-file.yaml")
+    val conf1 = baseConf.clone().set(templateFileConf.key, "template-file.yaml")
       .set(Config.KUBERNETES_SCHEDULER_NAME.key, "custom")
-    val pod = buildPod(sparkConf, client)
-    assert(pod.pod.getSpec.getSchedulerName === "custom")
+    val pod1 = buildPod(conf1, client)
+    assert(pod1.pod.getSpec.getSchedulerName === "custom")
+
+    val conf2 = baseConf.clone().set(templateFileConf.key, "template-file.yaml")
+      .set(Config.KUBERNETES_SCHEDULER_NAME.key, "custom")
+      .set(roleSpecificSchedulerNameConf.key, "rolescheduler")
+    val pod2 = buildPod(conf2, client)
+    assert(pod2.pod.getSpec.getSchedulerName === "rolescheduler")
   }
 
   test("load pod template if specified") {

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilderSuite.scala

@@ -34,6 +34,10 @@ class KubernetesDriverBuilderSuite extends PodBuilderSuite {
     Config.KUBERNETES_DRIVER_PODTEMPLATE_FILE
   }
 
+  override protected def roleSpecificSchedulerNameConf: ConfigEntry[_] = {
+    Config.KUBERNETES_DRIVER_SCHEDULER_NAME
+  }
+
   override protected def userFeatureStepsConf: ConfigEntry[_] = {
     Config.KUBERNETES_DRIVER_POD_FEATURE_STEPS
   }

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

@@ -33,6 +33,10 @@ class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
     Config.KUBERNETES_EXECUTOR_PODTEMPLATE_FILE
   }
 
+  override protected def roleSpecificSchedulerNameConf: ConfigEntry[_] = {
+    Config.KUBERNETES_EXECUTOR_SCHEDULER_NAME
+  }
+
   override protected def userFeatureStepsConf: ConfigEntry[_] = {
     Config.KUBERNETES_EXECUTOR_POD_FEATURE_STEPS
   }