restructured step based pattern to resolve comments

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

@@ -67,15 +67,14 @@ private[spark] case class KubernetesConf[T <: KubernetesRoleSpecificConf](
     .map(str => str.split(",").toSeq)
     .getOrElse(Seq.empty[String])
 
+  def getRoleConf: T = roleSpecificConf
+
   def pyFiles(): Option[String] = sparkConf
     .get(KUBERNETES_PYSPARK_PY_FILES)
 
   def pySparkMainResource(): Option[String] = sparkConf
     .get(KUBERNETES_PYSPARK_MAIN_APP_RESOURCE)
 
-  def pySparkAppArgs(): Option[String] = sparkConf
-    .get(KUBERNETES_PYSPARK_APP_ARGS)
-
   def pySparkPythonVersion(): String = sparkConf
       .get(PYSPARK_PYTHON_VERSION)
 
@@ -132,7 +131,6 @@ private[spark] object KubernetesConf {
               maybePyFiles.foreach{maybePyFiles =>
                 additionalFiles.appendAll(maybePyFiles.split(","))}
               sparkConfWithMainAppJar.set(KUBERNETES_PYSPARK_MAIN_APP_RESOURCE, res)
-              sparkConfWithMainAppJar.set(KUBERNETES_PYSPARK_APP_ARGS, appArgs.mkString(" "))
           }
           sparkConfWithMainAppJar.setIfMissing(MEMORY_OVERHEAD_FACTOR, 0.4)
     }

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

@@ -27,7 +27,6 @@ import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.submit._
 import org.apache.spark.internal.config._
-import org.apache.spark.launcher.SparkLauncher
 
 private[spark] class BasicDriverFeatureStep(
     conf: KubernetesConf[KubernetesDriverSpecificConf])
@@ -77,7 +76,7 @@ private[spark] class BasicDriverFeatureStep(
       ("cpu", new QuantityBuilder(false).withAmount(limitCores).build())
     }
 
-    val withoutArgsDriverContainer: ContainerBuilder = new ContainerBuilder(pod.container)
+    val driverContainer = new ContainerBuilder(pod.container)
       .withName(DRIVER_CONTAINER_NAME)
       .withImage(driverContainerImage)
       .withImagePullPolicy(conf.imagePullPolicy())
@@ -97,19 +96,8 @@ private[spark] class BasicDriverFeatureStep(
       .addToArgs(driverDockerContainer)
       .addToArgs("--properties-file", SPARK_CONF_PATH)
       .addToArgs("--class", conf.roleSpecificConf.mainClass)
+      .build()
 
-    val driverContainer =
-      if (driverDockerContainer == "driver-py") {
-        withoutArgsDriverContainer
-          .build()
-      } else {
-        // The user application jar is merged into the spark.jars list and managed through that
-        // property, so there is no need to reference it explicitly here.
-        withoutArgsDriverContainer
-          .addToArgs(SparkLauncher.NO_RESOURCE)
-          .addToArgs(conf.roleSpecificConf.appArgs: _*)
-          .build()
-      }
     val driverPod = new PodBuilder(pod.pod)
       .editOrNewMetadata()
         .withName(driverPodName)

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

@@ -0,0 +1,42 @@
+/*
+ * 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.features.bindings
+
+import io.fabric8.kubernetes.api.model.ContainerBuilder
+import io.fabric8.kubernetes.api.model.HasMetadata
+
+import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
+import org.apache.spark.deploy.k8s.KubernetesDriverSpecificConf
+import org.apache.spark.deploy.k8s.features.KubernetesFeatureConfigStep
+import org.apache.spark.launcher.SparkLauncher
+
+private[spark] class JavaDriverFeatureStep(
+  kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf])
+  extends KubernetesFeatureConfigStep {
+  override def configurePod(pod: SparkPod): SparkPod = {
+    val withDriverArgs = new ContainerBuilder(pod.container)
+      // The user application jar is merged into the spark.jars list and managed through that
+      // property, so there is no need to reference it explicitly here.
+      .addToArgs(SparkLauncher.NO_RESOURCE)
+      .addToArgs(kubernetesConf.roleSpecificConf.appArgs: _*)
+      .build()
+    SparkPod(pod.pod, withDriverArgs)
+  }
+  override def getAdditionalPodSystemProperties(): Map[String, String] = Map.empty
+
+  override def getAdditionalKubernetesResources(): Seq[HasMetadata] = Seq.empty
+}

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

@@ -16,42 +16,55 @@
  */
 package org.apache.spark.deploy.k8s.features.bindings
 
+import scala.collection.JavaConverters._
+
 import io.fabric8.kubernetes.api.model.ContainerBuilder
+import io.fabric8.kubernetes.api.model.EnvVar
+import io.fabric8.kubernetes.api.model.EnvVarBuilder
 import io.fabric8.kubernetes.api.model.HasMetadata
 
-import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesRoleSpecificConf, SparkPod}
+import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.deploy.k8s.KubernetesDriverSpecificConf
 import org.apache.spark.deploy.k8s.KubernetesUtils
 import org.apache.spark.deploy.k8s.features.KubernetesFeatureConfigStep
 
 private[spark] class PythonDriverFeatureStep(
-  kubernetesConf: KubernetesConf[_ <: KubernetesRoleSpecificConf])
+  kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf])
   extends KubernetesFeatureConfigStep {
   override def configurePod(pod: SparkPod): SparkPod = {
-    val mainResource = kubernetesConf.pySparkMainResource()
-    require(mainResource.isDefined, "PySpark Main Resource must be defined")
-    val otherPyFiles = kubernetesConf.pyFiles().map(pyFile =>
-      KubernetesUtils.resolveFileUrisAndPath(pyFile.split(","))
-        .mkString(":")).getOrElse("")
-    val withPythonPrimaryFileContainer = new ContainerBuilder(pod.container)
-      .addNewEnv()
-        .withName(ENV_PYSPARK_ARGS)
-        .withValue(kubernetesConf.pySparkAppArgs().getOrElse(""))
-        .endEnv()
-      .addNewEnv()
-        .withName(ENV_PYSPARK_PRIMARY)
-        .withValue(KubernetesUtils.resolveFileUri(mainResource.get))
-        .endEnv()
-      .addNewEnv()
-        .withName(ENV_PYSPARK_FILES)
-        .withValue(if (otherPyFiles == "") {""} else otherPyFiles)
-        .endEnv()
-      .addNewEnv()
-        .withName(ENV_PYSPARK_PYTHON_VERSION)
-        .withValue(kubernetesConf.pySparkPythonVersion())
-        .endEnv()
-      .build()
-    SparkPod(pod.pod, withPythonPrimaryFileContainer)
+    val roleConf = kubernetesConf.roleSpecificConf
+    require(roleConf.mainAppResource.isDefined, "PySpark Main Resource must be defined")
+    val maybePythonArgs: Option[EnvVar] = Option(roleConf.appArgs).filter(_.nonEmpty).map(
+      s =>
+        new EnvVarBuilder()
+          .withName(ENV_PYSPARK_ARGS)
+          .withValue(s.mkString(","))
+          .build())
+    val maybePythonFiles: Option[EnvVar] = kubernetesConf.pyFiles().map(
+      pyFiles =>
+        new EnvVarBuilder()
+          .withName(ENV_PYSPARK_FILES)
+          .withValue(KubernetesUtils.resolveFileUrisAndPath(pyFiles.split(","))
+            .mkString(":"))
+          .build())
+    val envSeq : Seq[EnvVar] =
+      Seq(new EnvVarBuilder()
+          .withName(ENV_PYSPARK_PRIMARY)
+          .withValue(KubernetesUtils.resolveFileUri(kubernetesConf.pySparkMainResource().get))
+        .build(),
+          new EnvVarBuilder()
+          .withName(ENV_PYSPARK_PYTHON_VERSION)
+          .withValue(kubernetesConf.pySparkPythonVersion())
+        .build())
+    val pythonEnvs = envSeq ++
+      maybePythonArgs.toSeq ++
+      maybePythonFiles.toSeq
+
+    val withPythonPrimaryContainer = new ContainerBuilder(pod.container)
+        .addAllToEnv(pythonEnvs.asJava).build()
+
+    SparkPod(pod.pod, withPythonPrimaryContainer)
   }
   override def getAdditionalPodSystemProperties(): Map[String, String] = Map.empty
 

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

@@ -214,7 +214,7 @@ private[spark] class KubernetesClientApplication extends SparkApplication {
     val kubernetesResourceNamePrefix = {
       s"$appName-$launchTime".toLowerCase.replaceAll("\\.", "-")
     }
-    sparkConf.set("spark.kubernetes.python.pyFiles", clientArguments.maybePyFiles.getOrElse(""))
+    sparkConf.set(KUBERNETES_PYSPARK_PY_FILES, clientArguments.maybePyFiles.getOrElse(""))
     val kubernetesConf = KubernetesConf.createDriverConf(
       sparkConf,
       appName,

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

@@ -19,7 +19,7 @@ package org.apache.spark.deploy.k8s.submit
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpec, KubernetesDriverSpecificConf, KubernetesRoleSpecificConf}
 import org.apache.spark.deploy.k8s.features.{BasicDriverFeatureStep, DriverKubernetesCredentialsFeatureStep, DriverServiceFeatureStep, MountSecretsFeatureStep}
 import org.apache.spark.deploy.k8s.features.KubernetesFeatureConfigStep
-import org.apache.spark.deploy.k8s.features.bindings.PythonDriverFeatureStep
+import org.apache.spark.deploy.k8s.features.bindings.{JavaDriverFeatureStep, PythonDriverFeatureStep}
 
 private[spark] class KubernetesDriverBuilder(
     provideBasicStep: (KubernetesConf[KubernetesDriverSpecificConf]) => BasicDriverFeatureStep =
@@ -32,8 +32,12 @@ private[spark] class KubernetesDriverBuilder(
     provideSecretsStep: (KubernetesConf[_ <: KubernetesRoleSpecificConf]
       => MountSecretsFeatureStep) =
       new MountSecretsFeatureStep(_),
+    provideJavaStep: (
+      KubernetesConf[KubernetesDriverSpecificConf]
+        => JavaDriverFeatureStep) =
+    new JavaDriverFeatureStep(_),
     providePythonStep: (
-      KubernetesConf[_ <: KubernetesRoleSpecificConf]
+      KubernetesConf[KubernetesDriverSpecificConf]
       => PythonDriverFeatureStep) =
       new PythonDriverFeatureStep(_)) {
 
@@ -45,16 +49,18 @@ private[spark] class KubernetesDriverBuilder(
       provideServiceStep(kubernetesConf))
     val maybeRoleSecretNamesStep = if (kubernetesConf.roleSecretNamesToMountPaths.nonEmpty) {
       Some(provideSecretsStep(kubernetesConf)) } else None
-    val maybeNonJVMBindings = kubernetesConf.roleSpecificConf.mainAppResource.getOrElse(None)
+    val bindingsStep = kubernetesConf.roleSpecificConf.mainAppResource.getOrElse(None)
       match {
+        case JavaMainAppResource(_) =>
+          Some(provideJavaStep(kubernetesConf))
         case PythonMainAppResource(_) =>
           Some(providePythonStep(kubernetesConf))
         case _ => None
     }
     val allFeatures: Seq[KubernetesFeatureConfigStep] =
       baseFeatures ++
       maybeRoleSecretNamesStep.toSeq ++
-      maybeNonJVMBindings.toSeq
+        bindingsStep.toSeq
     var spec = KubernetesDriverSpec.initialSpec(kubernetesConf.sparkConf.getAll.toMap)
     for (feature <- allFeatures) {
       val configuredPod = feature.configurePod(spec.pod)

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

@@ -0,0 +1,54 @@
+/*
+ * 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.features.bindings
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpecificConf, SparkPod}
+import org.apache.spark.deploy.k8s.submit.PythonMainAppResource
+
+class JavaDriverFeatureStepSuite extends SparkFunSuite {
+
+
+  test("Python Step modifies container correctly") {
+    val baseDriverPod = SparkPod.initialPod()
+    val sparkConf = new SparkConf(false)
+    val kubernetesConf = KubernetesConf(
+      sparkConf,
+      KubernetesDriverSpecificConf(
+        Some(PythonMainAppResource("local:///main.jar")),
+        "test-app",
+        "java-runner",
+        Seq("5 7")),
+      "",
+      "",
+      Map.empty,
+      Map.empty,
+      Map.empty,
+      Map.empty,
+      Seq.empty[String])
+
+    val step = new JavaDriverFeatureStep(kubernetesConf)
+    val driverPod = step.configurePod(baseDriverPod).pod
+    val driverContainerwithJavaStep = step.configurePod(baseDriverPod).container
+    assert(driverContainerwithJavaStep.getArgs.size === 2)
+    val args = driverContainerwithJavaStep
+      .getArgs.asScala
+    assert(args === List("spark-internal", "5 7"))
+  }
+}

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

@@ -38,14 +38,14 @@ class PythonDriverFeatureStepSuite extends SparkFunSuite {
       .set(KUBERNETES_PYSPARK_MAIN_APP_RESOURCE, mainResource)
       .set(KUBERNETES_PYSPARK_PY_FILES, pyFiles.mkString(","))
       .set("spark.files", "local:///example.py")
-      .set(KUBERNETES_PYSPARK_APP_ARGS, "5 7")
+      .set(PYSPARK_PYTHON_VERSION, "2")
     val kubernetesConf = KubernetesConf(
       sparkConf,
       KubernetesDriverSpecificConf(
         Some(PythonMainAppResource("local:///main.py")),
         "test-app",
         "python-runner",
-        Seq.empty[String]),
+        Seq("5 7")),
       "",
       "",
       Map.empty,
@@ -57,7 +57,7 @@ class PythonDriverFeatureStepSuite extends SparkFunSuite {
     val step = new PythonDriverFeatureStep(kubernetesConf)
     val driverPod = step.configurePod(baseDriverPod).pod
     val driverContainerwithPySpark = step.configurePod(baseDriverPod).container
-    assert(driverContainerwithPySpark.getEnv.size === 4)
+//    assert(driverContainerwithPySpark.getEnv.size === 4)
     val envs = driverContainerwithPySpark
       .getEnv
       .asScala
@@ -91,14 +91,12 @@ class PythonDriverFeatureStepSuite extends SparkFunSuite {
     val step = new PythonDriverFeatureStep(kubernetesConf)
     val driverPod = step.configurePod(baseDriverPod).pod
     val driverContainerwithPySpark = step.configurePod(baseDriverPod).container
-    assert(driverContainerwithPySpark.getEnv.size === 4)
+    assert(driverContainerwithPySpark.getEnv.size === 2)
     val envs = driverContainerwithPySpark
       .getEnv
       .asScala
       .map(env => (env.getName, env.getValue))
       .toMap
-    assert(envs(ENV_PYSPARK_FILES) === "")
-    assert(envs(ENV_PYSPARK_ARGS) === "")
     assert(envs(ENV_PYSPARK_PYTHON_VERSION) === "3")
   }
 }