Add tests and refactor KubernetesVolumeUtils

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

@@ -33,15 +33,18 @@ private[spark] object KubernetesVolumeUtils {
     prefix: String): Iterable[KubernetesVolumeSpec] = {
     val properties = sparkConf.getAllWithPrefix(prefix)
 
-    val propsByTypeName = properties.map { case (k, v) =>
-      k.split('.').toSeq match {
-        case tpe :: name :: rest => ((tpe, name), (rest.mkString("."), v))
-      }
-    }.groupBy(_._1).mapValues(_.map(_._2))
+    val propsByTypeName: Map[(String, String), Array[(String, String)]] =
+      properties.flatMap { case (k, v) =>
+        k.split('.').toList match {
+          case tpe :: name :: rest => Some(((tpe, name), (rest.mkString("."), v)))
+          case _ => None
+        }
+      }.groupBy(_._1).mapValues(_.map(_._2))
 
     propsByTypeName.map { case ((tpe, name), props) =>
-      val mountProps = props.filter(_._1.startsWith(KUBERNETES_VOLUMES_MOUNT_KEY)).toMap
-      val options = props.filter(_._1.startsWith(KUBERNETES_VOLUMES_OPTIONS_KEY)).toMap
+      val propMap = props.toMap
+      val mountProps = getAllWithPrefix(propMap, s"$KUBERNETES_VOLUMES_MOUNT_KEY.")
+      val options = getAllWithPrefix(propMap, s"$KUBERNETES_VOLUMES_OPTIONS_KEY.")
 
       KubernetesVolumeSpec(
         volumeName = name,
@@ -53,4 +56,16 @@ private[spark] object KubernetesVolumeUtils {
     }
   }
 
+  /**
+   * Extract subset of elements with keys matching on prefix, which is then excluded
+   * @param props properties to extract data from
+   * @param prefix prefix to match against
+   * @return subset of original props
+   */
+  private def getAllWithPrefix(props: Map[String, String], prefix: String): Map[String, String] = {
+    props
+      .filterKeys(_.startsWith(prefix))
+      .map { case (k, v) => k.substring(prefix.length) -> v }
+  }
+
 }

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

@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+package org.apache.spark.deploy.k8s
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+
+class KubernetesVolumeUtilsSuite extends SparkFunSuite {
+  test("Parses volume options correctly") {
+    val sparkConf = new SparkConf(false)
+    sparkConf.set("test.volumeType.volumeName.mount.path", "/path")
+    sparkConf.set("test.volumeType.volumeName.mount.readOnly", "true")
+    sparkConf.set("test.volumeType.volumeName.options.option1", "value1")
+    sparkConf.set("test.volumeType.volumeName.options.option2", "value2")
+
+    val volumeSpec = KubernetesVolumeUtils.parseVolumesWithPrefix(sparkConf, "test.").head
+    assert(volumeSpec.volumeName === "volumeName")
+    assert(volumeSpec.volumeType === "volumeType")
+    assert(volumeSpec.mountPath === "/path")
+    assert(volumeSpec.mountReadOnly === true)
+    assert(volumeSpec.optionsSpec === Map("option1" -> "value1", "option2" -> "value2"))
+  }
+}