[SPARK-48017] Add Spark application submission worker for operator

build.gradle

@@ -25,6 +25,11 @@ subprojects {
 
   repositories {
     mavenCentral()
+    // This is a workaround to resolve Spark 4.0.0-preview-1
+    // To be removed for official release
+    maven {
+      url "https://repository.apache.org/content/repositories/orgapachespark-1454/"
+    }
   }
 
   apply plugin: 'checkstyle'

gradle.properties

@@ -18,17 +18,23 @@
 group=org.apache.spark.k8s.operator
 version=0.1.0
 
+# Caution: fabric8 version should be aligned with Spark dependency
 fabric8Version=6.12.1
 commonsLang3Version=3.14.0
 commonsIOVersion=2.16.1
 lombokVersion=1.18.32
 
+#Spark
+scalaVersion=2.13
+sparkVersion=4.0.0-preview1
+
 # Logging
 log4jVersion=2.22.1
 
 # Test
 junitVersion=5.10.2
 jacocoVersion=0.8.12
+mockitoVersion=5.11.0
 
 # Build Analysis
 checkstyleVersion=10.15.0

settings.gradle

@@ -1,2 +1,3 @@
 rootProject.name = 'apache-spark-kubernetes-operator'
 include 'spark-operator-api'
+include 'spark-submission-worker'

spark-operator-api/build.gradle

@@ -18,6 +18,7 @@ dependencies {
 
   testImplementation platform("org.junit:junit-bom:$junitVersion")
   testImplementation 'org.junit.jupiter:junit-jupiter'
+  testRuntimeOnly "org.junit.platform:junit-platform-launcher"
 }
 
 test {

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/utils/ModelUtils.java

@@ -36,6 +36,7 @@
 import io.fabric8.kubernetes.api.model.PodTemplateSpec;
 import org.apache.commons.lang3.StringUtils;
 
+import org.apache.spark.k8s.operator.SparkApplication;
 import org.apache.spark.k8s.operator.spec.ApplicationSpec;
 
 public class ModelUtils {
@@ -107,4 +108,12 @@ public static boolean overrideExecutorTemplateEnabled(ApplicationSpec applicatio
         && applicationSpec.getExecutorSpec() != null
         && applicationSpec.getExecutorSpec().getPodTemplateSpec() != null;
   }
+
+  public static long getAttemptId(final SparkApplication app) {
+    long attemptId = 0L;
+    if (app.getStatus() != null && app.getStatus().getCurrentAttemptSummary() != null) {
+      attemptId = app.getStatus().getCurrentAttemptSummary().getAttemptInfo().getId();
+    }
+    return attemptId;
+  }
 }

spark-submission-worker/build.gradle

@@ -0,0 +1,18 @@
+dependencies {
+  implementation project(":spark-operator-api")
+
+  implementation("org.apache.spark:spark-kubernetes_$scalaVersion:$sparkVersion")
+
+  compileOnly("org.projectlombok:lombok:$lombokVersion")
+  annotationProcessor("org.projectlombok:lombok:$lombokVersion")
+
+  testImplementation platform("org.junit:junit-bom:$junitVersion")
+  testImplementation "org.mockito:mockito-core:$mockitoVersion"
+  testImplementation "org.junit.jupiter:junit-jupiter:$junitVersion"
+  testImplementation "io.fabric8:kubernetes-server-mock:$fabric8Version"
+  testRuntimeOnly "org.junit.platform:junit-platform-launcher"
+}
+
+test {
+  useJUnitPlatform()
+}

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppDriverConf.java

@@ -0,0 +1,64 @@
+/*
+ * 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.k8s.operator;
+
+import scala.Option;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.deploy.k8s.Config;
+import org.apache.spark.deploy.k8s.KubernetesDriverConf;
+import org.apache.spark.deploy.k8s.KubernetesVolumeUtils;
+import org.apache.spark.deploy.k8s.submit.KubernetesClientUtils;
+import org.apache.spark.deploy.k8s.submit.MainAppResource;
+
+public class SparkAppDriverConf extends KubernetesDriverConf {
+  private SparkAppDriverConf(
+      SparkConf sparkConf,
+      String appId,
+      MainAppResource mainAppResource,
+      String mainClass,
+      String[] appArgs,
+      Option<String> proxyUser) {
+    super(sparkConf, appId, mainAppResource, mainClass, appArgs, proxyUser, null);
+  }
+
+  public static SparkAppDriverConf create(
+      SparkConf sparkConf,
+      String appId,
+      MainAppResource mainAppResource,
+      String mainClass,
+      String[] appArgs,
+      Option<String> proxyUser) {
+    // pre-create check only
+    KubernetesVolumeUtils.parseVolumesWithPrefix(
+        sparkConf, Config.KUBERNETES_EXECUTOR_VOLUMES_PREFIX());
+    return new SparkAppDriverConf(sparkConf, appId, mainAppResource, mainClass, appArgs, proxyUser);
+  }
+
+  /** Application managed by operator has a deterministic prefix */
+  @Override
+  public String resourceNamePrefix() {
+    return appId();
+  }
+
+  public String configMapNameDriver() {
+    return KubernetesClientUtils.configMapName(String.format("spark-drv-%s", resourceNamePrefix()));
+  }
+}

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java

@@ -0,0 +1,129 @@
+/*
+ * 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.k8s.operator;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import scala.Tuple2;
+import scala.collection.immutable.HashMap;
+import scala.collection.immutable.Map;
+import scala.jdk.CollectionConverters;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import lombok.Getter;
+import org.apache.commons.lang3.StringUtils;
+
+import org.apache.spark.deploy.k8s.Config;
+import org.apache.spark.deploy.k8s.Constants;
+import org.apache.spark.deploy.k8s.KubernetesDriverSpec;
+import org.apache.spark.deploy.k8s.SparkPod;
+import org.apache.spark.deploy.k8s.submit.KubernetesClientUtils;
+
+/**
+ * Resembles resources that would be directly launched by operator. Based on resolved
+ * org.apache.spark.deploy.k8s.KubernetesDriverSpec, it:
+ *
+ * <ul>
+ *   <li>Add ConfigMap as a resource for driver
+ *   <li>Converts scala types to Java for easier reference from operator
+ * </ul>
+ *
+ * <p>This is not thread safe and not expected to be shared among reconciler threads
+ */
+public class SparkAppResourceSpec {
+  @Getter private final Pod configuredPod;
+  @Getter private final List<HasMetadata> driverPreResources;
+  @Getter private final List<HasMetadata> driverResources;
+  private final SparkAppDriverConf kubernetesDriverConf;
+
+  public SparkAppResourceSpec(
+      SparkAppDriverConf kubernetesDriverConf, KubernetesDriverSpec kubernetesDriverSpec) {
+    this.kubernetesDriverConf = kubernetesDriverConf;
+    String namespace = kubernetesDriverConf.sparkConf().get(Config.KUBERNETES_NAMESPACE().key());
+    Map<String, String> confFilesMap =
+        KubernetesClientUtils.buildSparkConfDirFilesMap(
+                kubernetesDriverConf.configMapNameDriver(),
+                kubernetesDriverConf.sparkConf(),
+                kubernetesDriverSpec.systemProperties())
+            .$plus(new Tuple2<>(Config.KUBERNETES_NAMESPACE().key(), namespace));
+    SparkPod sparkPod = addConfigMap(kubernetesDriverSpec.pod(), confFilesMap);
+    this.configuredPod =
+        new PodBuilder(sparkPod.pod())
+            .editSpec()
+            .addToContainers(sparkPod.container())
+            .endSpec()
+            .build();
+    this.driverPreResources =
+        new ArrayList<>(
+            CollectionConverters.SeqHasAsJava(kubernetesDriverSpec.driverPreKubernetesResources())
+                .asJava());
+    this.driverResources =
+        new ArrayList<>(
+            CollectionConverters.SeqHasAsJava(kubernetesDriverSpec.driverKubernetesResources())
+                .asJava());
+    this.driverResources.add(
+        KubernetesClientUtils.buildConfigMap(
+            kubernetesDriverConf.configMapNameDriver(), confFilesMap, new HashMap<>()));
+    this.driverPreResources.forEach(r -> setNamespaceIfMissing(r, namespace));
+    this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));
+  }
+
+  private void setNamespaceIfMissing(HasMetadata resource, String namespace) {
+    if (StringUtils.isNotEmpty(resource.getMetadata().getNamespace())) {
+      return;
+    }
+    resource.getMetadata().setNamespace(namespace);
+  }
+
+  private SparkPod addConfigMap(SparkPod pod, Map<String, String> confFilesMap) {
+    Container containerWithVolume =
+        new ContainerBuilder(pod.container())
+            .addNewEnv()
+            .withName(Constants.ENV_SPARK_CONF_DIR())
+            .withValue(Constants.SPARK_CONF_DIR_INTERNAL())
+            .endEnv()
+            .addNewVolumeMount()
+            .withName(Constants.SPARK_CONF_VOLUME_DRIVER())
+            .withMountPath(Constants.SPARK_CONF_DIR_INTERNAL())
+            .endVolumeMount()
+            .build();
+    Pod podWithVolume =
+        new PodBuilder(pod.pod())
+            .editSpec()
+            .addNewVolume()
+            .withName(Constants.SPARK_CONF_VOLUME_DRIVER())
+            .withNewConfigMap()
+            .withItems(
+                CollectionConverters.SeqHasAsJava(
+                        KubernetesClientUtils.buildKeyToPathObjects(confFilesMap))
+                    .asJava())
+            .withName(kubernetesDriverConf.configMapNameDriver())
+            .endConfigMap()
+            .endVolume()
+            .endSpec()
+            .build();
+    return new SparkPod(podWithVolume, containerWithVolume);
+  }
+}