[SPARK-38480][K8S] Remove spark.kubernetes.job.queue in favor of spark.kubernetes.driver.podGroupTemplateFile

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

@@ -306,13 +306,6 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
-  val KUBERNETES_JOB_QUEUE = ConfigBuilder("spark.kubernetes.job.queue")
-    .doc("The name of the queue to which the job is submitted. This info " +
-      "will be stored in configuration and passed to specific feature step.")
-    .version("3.3.0")
-    .stringConf
-    .createOptional
-
   val KUBERNETES_EXECUTOR_REQUEST_CORES =
     ConfigBuilder("spark.kubernetes.executor.request.cores")
       .doc("Specify the cpu request for each executor pod")

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

@@ -32,7 +32,6 @@ private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureCon
 
   private lazy val podGroupName = s"${kubernetesConf.appId}-podgroup"
   private lazy val namespace = kubernetesConf.namespace
-  private lazy val queue = kubernetesConf.get(KUBERNETES_JOB_QUEUE)
   private var priorityClassName: Option[String] = None
 
   override def init(config: KubernetesDriverConf): Unit = {
@@ -60,7 +59,6 @@ private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureCon
 
     var spec = pg.getSpec
     if (spec == null) spec = new PodGroupSpec
-    queue.foreach(spec.setQueue(_))
     priorityClassName.foreach(spec.setPriorityClassName(_))
     pg.setSpec(spec)
 

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

@@ -41,16 +41,6 @@ class VolcanoFeatureStepSuite extends SparkFunSuite {
     assert(podGroup.getMetadata.getName === s"${kubernetesConf.appId}-podgroup")
   }
 
-  test("SPARK-38818: Support `spark.kubernetes.job.queue`") {
-    val sparkConf = new SparkConf()
-      .set(KUBERNETES_JOB_QUEUE.key, "queue1")
-    val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf)
-    val step = new VolcanoFeatureStep()
-    step.init(kubernetesConf)
-    val podGroup = step.getAdditionalPreKubernetesResources().head.asInstanceOf[PodGroup]
-    assert(podGroup.getSpec.getQueue === "queue1")
-  }
-
   test("SPARK-36061: Executor Pod with Volcano PodGroup") {
     val sparkConf = new SparkConf()
     val kubernetesConf = KubernetesTestConf.createExecutorConf(sparkConf)

resource-managers/kubernetes/integration-tests/src/test/resources/volcano/queue-driver-podgroup-template.yml

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+apiVersion: scheduling.volcano.sh/v1beta1
+kind: PodGroup
+spec:
+  queue: queue

resource-managers/kubernetes/integration-tests/src/test/resources/volcano/queue0-driver-podgroup-template.yml

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+apiVersion: scheduling.volcano.sh/v1beta1
+kind: PodGroup
+spec:
+  queue: queue0

resource-managers/kubernetes/integration-tests/src/test/resources/volcano/queue1-driver-podgroup-template.yml

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+apiVersion: scheduling.volcano.sh/v1beta1
+kind: PodGroup
+spec:
+  queue: queue1

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/VolcanoTestsSuite.scala

@@ -200,7 +200,7 @@ private[spark] trait VolcanoTestsSuite extends BeforeAndAfterEach { k8sSuite: Ku
       groupLoc: Option[String] = None,
       queue: Option[String] = None,
       driverTemplate: Option[String] = None): SparkAppConf = {
-    val conf = kubernetesTestComponents.newSparkAppConf()
+    var conf = kubernetesTestComponents.newSparkAppConf()
       .set(CONTAINER_IMAGE.key, image)
       .set(KUBERNETES_DRIVER_POD_NAME.key, driverPodName)
       .set(s"${KUBERNETES_DRIVER_LABEL_PREFIX}spark-app-locator", appLoc)
@@ -210,7 +210,12 @@ private[spark] trait VolcanoTestsSuite extends BeforeAndAfterEach { k8sSuite: Ku
       .set(KUBERNETES_SCHEDULER_NAME.key, "volcano")
       .set(KUBERNETES_DRIVER_POD_FEATURE_STEPS.key, VOLCANO_FEATURE_STEP)
       .set(KUBERNETES_EXECUTOR_POD_FEATURE_STEPS.key, VOLCANO_FEATURE_STEP)
-    queue.foreach(conf.set(KUBERNETES_JOB_QUEUE.key, _))
+    queue.foreach { q =>
+      conf = conf.set(KUBERNETES_DRIVER_PODGROUP_TEMPLATE_FILE.key,
+        new File(
+          getClass.getResource(s"/volcano/$q-driver-podgroup-template.yml").getFile
+        ).getAbsolutePath)
+    }
     groupLoc.foreach { locator =>
       conf.set(s"${KUBERNETES_DRIVER_LABEL_PREFIX}spark-group-locator", locator)
       conf.set(s"${KUBERNETES_EXECUTOR_LABEL_PREFIX}spark-group-locator", locator)