[SPARK-52997] Fixes wrong worker assignment if multiple clusters are deployed to the same namespace

### What changes were proposed in this pull request?
Updated the podSelector for master and worker services to include both clusterRole and name labels.

### Why are the changes needed?
Using only clusterRole caused service misrouting when multiple Spark clusters were deployed in the same namespace. Adding name ensures correct pod targeting.

### Does this PR introduce any user-facing change?
No, this is an internal fix to service selectors.

### How was this patch tested?
Tested with multiple clusters in the same namespace. Verified each service only matched its own pods via kubectl describe service.
Also adapted unit tests to reflect new behaviour

### Was this patch authored or co-authored using generative AI tooling?
Yes, PR metadata was assisted by AI, but code changes were made manually.

Closes apache#291 from schmaxXximilian/main.

Authored-by: Schmöller Maximilian <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 7915164)

Author: schmaxXximilian

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

@@ -117,7 +117,8 @@ private static Service buildMasterService(
         .endMetadata()
         .withNewSpecLike(serviceSpec)
         .withClusterIP("None")
-        .withSelector(
+        .addToSelector(Collections.singletonMap(LABEL_SPARK_CLUSTER_NAME, name))
+        .addToSelector(
             Collections.singletonMap(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE))
         .addNewPort()
         .withName("web")
@@ -149,7 +150,8 @@ private static Service buildWorkerService(
         .endMetadata()
         .withNewSpecLike(serviceSpec)
         .withClusterIP("None")
-        .withSelector(
+        .addToSelector(Collections.singletonMap(LABEL_SPARK_CLUSTER_NAME, name))
+        .addToSelector(
             Collections.singletonMap(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE))
         .addNewPort()
         .withName("web")
@@ -186,6 +188,7 @@ private static StatefulSet buildMasterStatefulSet(
             .editOrNewTemplate()
             .editOrNewMetadata()
             .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
+            .addToLabels(LABEL_SPARK_CLUSTER_NAME, name)
             .addToLabels(LABEL_SPARK_VERSION_NAME, version)
             .endMetadata()
             .editOrNewSpec()
@@ -253,6 +256,7 @@ private static StatefulSet buildWorkerStatefulSet(
             .editOrNewTemplate()
             .editOrNewMetadata()
             .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE)
+            .addToLabels(LABEL_SPARK_CLUSTER_NAME, name)
             .addToLabels(LABEL_SPARK_VERSION_NAME, version)
             .endMetadata()
             .editOrNewSpec()

spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterResourceSpecTest.java

@@ -19,12 +19,13 @@
 
 package org.apache.spark.k8s.operator;
 
-import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_VERSION_NAME;
+import static org.apache.spark.k8s.operator.Constants.*;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.Map;
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -129,6 +130,13 @@ void testWorkerServiceWithTemplate() {
     assertEquals("bar", service1.getMetadata().getLabels().get("foo"));
     assertEquals("4.0.0", service1.getMetadata().getLabels().get(LABEL_SPARK_VERSION_NAME));
     assertEquals("foo", service1.getSpec().getExternalName());
+    assertEquals(
+        Map.of(
+            LABEL_SPARK_CLUSTER_NAME,
+            "cluster-name",
+            LABEL_SPARK_ROLE_NAME,
+            LABEL_SPARK_ROLE_WORKER_VALUE),
+        service1.getSpec().getSelector());
   }
 
   @Test
@@ -151,6 +159,13 @@ void testMasterServiceWithTemplate() {
     assertEquals("bar", service1.getMetadata().getLabels().get("foo"));
     assertEquals("4.0.0", service1.getMetadata().getLabels().get(LABEL_SPARK_VERSION_NAME));
     assertEquals("foo", service1.getSpec().getExternalName());
+    assertEquals(
+        Map.of(
+            LABEL_SPARK_CLUSTER_NAME,
+            "cluster-name",
+            LABEL_SPARK_ROLE_NAME,
+            LABEL_SPARK_ROLE_MASTER_VALUE),
+        service1.getSpec().getSelector());
   }
 
   @Test
@@ -172,6 +187,17 @@ void testMasterStatefulSet() {
     SparkClusterResourceSpec spec2 = new SparkClusterResourceSpec(cluster, sparkConf);
     StatefulSet statefulSet2 = spec2.getMasterStatefulSet();
     assertEquals("other-namespace", statefulSet2.getMetadata().getNamespace());
+    assertEquals(
+        "cluster-name",
+        statefulSet2
+            .getSpec()
+            .getTemplate()
+            .getMetadata()
+            .getLabels()
+            .get(LABEL_SPARK_CLUSTER_NAME));
+    assertEquals(
+        LABEL_SPARK_ROLE_MASTER_VALUE,
+        statefulSet2.getSpec().getTemplate().getMetadata().getLabels().get(LABEL_SPARK_ROLE_NAME));
   }
 
   @Test
@@ -236,6 +262,17 @@ void testWorkerStatefulSet() {
     SparkClusterResourceSpec spec2 = new SparkClusterResourceSpec(cluster, sparkConf);
     StatefulSet statefulSet2 = spec2.getWorkerStatefulSet();
     assertEquals("other-namespace", statefulSet2.getMetadata().getNamespace());
+    assertEquals(
+        "cluster-name",
+        statefulSet2
+            .getSpec()
+            .getTemplate()
+            .getMetadata()
+            .getLabels()
+            .get(LABEL_SPARK_CLUSTER_NAME));
+    assertEquals(
+        LABEL_SPARK_ROLE_WORKER_VALUE,
+        statefulSet2.getSpec().getTemplate().getMetadata().getLabels().get(LABEL_SPARK_ROLE_NAME));
   }
 
   @Test