[SPARK-49389] Support master|worker container templates

### What changes were proposed in this pull request?

This PR aims to support `master` container and `worker` container templates.

### Why are the changes needed?

Previously, pod template throws exceptions on containers whose name is `master` or `worker` because two `master|worker` containers were created. After this PR, the problem is resolved.
- apache#80

### Does this PR introduce _any_ user-facing change?

No, this is a new unreleased feature.

### How was this patch tested?

For now, we need to run manual tests.
```
$ ./gradlew build buildDockerImage spark-operator-api:relocateGeneratedCRD -x check
$ helm install spark-kubernetes-operator -f build-tools/helm/spark-kubernetes-operator/values.yaml build-tools/helm/spark-kubernetes-operator/
$ kubectl apply -f examples/cluster-with-template.yaml
$ kubectl get pod -l spark-role=master -oyaml | yq '.items[0].spec.containers[0].resources'
limits:
  cpu: "2"
  memory: 2Gi
requests:
  cpu: "2"
  memory: 2Gi
```

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#104 from dongjoon-hyun/SPARK-49389.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

examples/cluster-with-template.yaml

@@ -35,6 +35,14 @@ spec:
           securityContext:
             runAsUser: 0
           containers:
+          - name: master
+            resources:
+              requests:
+                cpu: "2"
+                memory: "2Gi"
+              limits:
+                cpu: "2"
+                memory: "2Gi"
           - name: sidecar
             image: registry.k8s.io/pause
             resources:
@@ -58,6 +66,14 @@ spec:
           securityContext:
             runAsUser: 0
           containers:
+          - name: worker
+            resources:
+              requests:
+                cpu: "2"
+                memory: "2Gi"
+              limits:
+                cpu: "2"
+                memory: "2Gi"
           - name: sidecar
             image: registry.k8s.io/pause
             resources:

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

@@ -153,27 +153,31 @@ private static StatefulSet buildMasterStatefulSet(
       String options,
       ObjectMeta objectMeta,
       StatefulSetSpec statefulSetSpec) {
-    return new StatefulSetBuilder()
-        .withNewMetadataLike(objectMeta)
-        .withName(name + "-master")
-        .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
-        .withNamespace(namespace)
-        .endMetadata()
-        .withNewSpecLike(statefulSetSpec)
-        .withPodManagementPolicy("Parallel")
-        .withReplicas(1)
-        .editOrNewSelector()
-        .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
-        .endSelector()
-        .editOrNewTemplate()
-        .editOrNewMetadata()
-        .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
-        .endMetadata()
-        .editOrNewSpec()
-        .withSchedulerName(scheduler)
-        .withTerminationGracePeriodSeconds(0L)
-        .addNewContainer()
-        .withName("master")
+    var partialStatefulSet =
+        new StatefulSetBuilder()
+            .withNewMetadataLike(objectMeta)
+            .withName(name + "-master")
+            .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
+            .withNamespace(namespace)
+            .endMetadata()
+            .withNewSpecLike(statefulSetSpec)
+            .withPodManagementPolicy("Parallel")
+            .withReplicas(1)
+            .editOrNewSelector()
+            .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
+            .endSelector()
+            .editOrNewTemplate()
+            .editOrNewMetadata()
+            .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
+            .endMetadata()
+            .editOrNewSpec()
+            .withSchedulerName(scheduler)
+            .withTerminationGracePeriodSeconds(0L);
+    if (!partialStatefulSet.hasMatchingContainer(p -> "master".equals(p.getName()))) {
+      partialStatefulSet = partialStatefulSet.addNewContainer().withName("master").endContainer();
+    }
+    return partialStatefulSet
+        .editMatchingContainer(p -> "master".equals(p.getName()))
         .withImage(image)
         .addNewEnv()
         .withName("SPARK_MASTER_OPTS")
@@ -212,31 +216,35 @@ private static StatefulSet buildWorkerStatefulSet(
       String options,
       ObjectMeta metadata,
       StatefulSetSpec statefulSetSpec) {
-    return new StatefulSetBuilder()
-        .withNewMetadataLike(metadata)
-        .withName(name + "-worker")
-        .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
-        .withNamespace(namespace)
-        .endMetadata()
-        .withNewSpecLike(statefulSetSpec)
-        .withPodManagementPolicy("Parallel")
-        .withReplicas(initWorkers)
-        .withServiceName(name + "-worker-svc")
-        .editOrNewSelector()
-        .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
-        .endSelector()
-        .editOrNewTemplate()
-        .editOrNewMetadata()
-        .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
-        .endMetadata()
-        .editOrNewSpec()
-        .withSchedulerName(scheduler)
-        .withTerminationGracePeriodSeconds(0L)
-        .withNewDnsConfig()
-        .withSearches(String.format("%s-worker-svc.%s.svc.cluster.local", name, namespace))
-        .endDnsConfig()
-        .addNewContainer()
-        .withName("worker")
+    var partialStatefulSet =
+        new StatefulSetBuilder()
+            .withNewMetadataLike(metadata)
+            .withName(name + "-worker")
+            .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
+            .withNamespace(namespace)
+            .endMetadata()
+            .withNewSpecLike(statefulSetSpec)
+            .withPodManagementPolicy("Parallel")
+            .withReplicas(initWorkers)
+            .withServiceName(name + "-worker-svc")
+            .editOrNewSelector()
+            .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
+            .endSelector()
+            .editOrNewTemplate()
+            .editOrNewMetadata()
+            .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
+            .endMetadata()
+            .editOrNewSpec()
+            .withSchedulerName(scheduler)
+            .withTerminationGracePeriodSeconds(0L)
+            .withNewDnsConfig()
+            .withSearches(String.format("%s-worker-svc.%s.svc.cluster.local", name, namespace))
+            .endDnsConfig();
+    if (!partialStatefulSet.hasMatchingContainer(p -> "worker".equals(p.getName()))) {
+      partialStatefulSet = partialStatefulSet.addNewContainer().withName("worker").endContainer();
+    }
+    return partialStatefulSet
+        .editMatchingContainer(p -> "worker".equals(p.getName()))
         .withImage(image)
         .addNewEnv()
         .withName("SPARK_LOG_DIR")