[SPARK-38379][K8S] Fix Kubernetes Client mode when mounting persisten…

…t volume with storage class ### What changes were proposed in this pull request? Running spark-shell in client mode on Kubernetes cluster when mounting persistent volumes with a storage class results in a big warning being thrown on startup. https://issues.apache.org/jira/browse/SPARK-38379 The issue here is there is a race condition between when spark.app.id is set in SparkContext and when its used, so change to use the KubernetesConf appId, which is what is used to set spark.app.id. ### Why are the changes needed? Throws big warning to user and I believe the label is wrong as well. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test added. The test fails without the fix. Also manually tested on real k8s cluster. Closes apache#35792 from tgravescs/fixVolk8s. Authored-by: Thomas Graves <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>
mantov5 · Mar 10, 2022 · f286416 · f286416
1 parent 82b6194
commit f286416
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 1 deletion.
diff --git a/...es/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountVolumesFeatureStep.scala b/...es/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountVolumesFeatureStep.scala
@@ -85,7 +85,7 @@ private[spark] class MountVolumesFeatureStep(conf: KubernetesConf)
               .withApiVersion("v1")
               .withNewMetadata()
                 .withName(claimName)
-                .addToLabels(SPARK_APP_ID_LABEL, conf.sparkConf.getAppId)
+                .addToLabels(SPARK_APP_ID_LABEL, conf.appId)
                 .endMetadata()
               .withNewSpec()
                 .withStorageClassName(storageClass.get)

diff --git a/...re/src/test/scala/org/apache/spark/deploy/k8s/features/MountVolumesFeatureStepSuite.scala b/...re/src/test/scala/org/apache/spark/deploy/k8s/features/MountVolumesFeatureStepSuite.scala
@@ -89,6 +89,31 @@ class MountVolumesFeatureStepSuite extends SparkFunSuite {
     assert(executorPVC.getClaimName === s"pvc-spark-${KubernetesTestConf.EXECUTOR_ID}")
   }
 
+  test("SPARK-32713 Mounts parameterized persistentVolumeClaims in executors with storage class") {
+    val volumeConf = KubernetesVolumeSpec(
+      "testVolume",
+      "/tmp",
+      "",
+      true,
+      KubernetesPVCVolumeConf("pvc-spark-SPARK_EXECUTOR_ID", Some("fast"), Some("512mb"))
+    )
+    val driverConf = KubernetesTestConf.createDriverConf(volumes = Seq(volumeConf))
+    val driverStep = new MountVolumesFeatureStep(driverConf)
+    val driverPod = driverStep.configurePod(SparkPod.initialPod())
+
+    assert(driverPod.pod.getSpec.getVolumes.size() === 1)
+    val driverPVC = driverPod.pod.getSpec.getVolumes.get(0).getPersistentVolumeClaim
+    assert(driverPVC.getClaimName === "pvc-spark-SPARK_EXECUTOR_ID")
+
+    val executorConf = KubernetesTestConf.createExecutorConf(volumes = Seq(volumeConf))
+    val executorStep = new MountVolumesFeatureStep(executorConf)
+    val executorPod = executorStep.configurePod(SparkPod.initialPod())
+
+    assert(executorPod.pod.getSpec.getVolumes.size() === 1)
+    val executorPVC = executorPod.pod.getSpec.getVolumes.get(0).getPersistentVolumeClaim
+    assert(executorPVC.getClaimName === s"pvc-spark-${KubernetesTestConf.EXECUTOR_ID}")
+  }
+
   test("Create and mounts persistentVolumeClaims in driver") {
     val volumeConf = KubernetesVolumeSpec(
       "testVolume",