fixes issue where multiple sparkclusters in one namespace will lead to wrong worker assignment

Author: schmaxXximilian

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

@@ -21,7 +21,7 @@
 
 import static org.apache.spark.k8s.operator.Constants.*;
 
-import java.util.Collections;
+import java.util.Map;
 import java.util.Optional;
 
 import scala.Tuple2;
@@ -117,8 +117,9 @@ private static Service buildMasterService(
         .endMetadata()
         .withNewSpecLike(serviceSpec)
         .withClusterIP("None")
+        .withClusterIP("None")
         .withSelector(
-            Collections.singletonMap(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE))
+            Map.of(LABEL_SPARK_CLUSTER_NAME, name, LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE))
         .addNewPort()
         .withName("web")
         .withPort(8080)
@@ -150,7 +151,7 @@ private static Service buildWorkerService(
         .withNewSpecLike(serviceSpec)
         .withClusterIP("None")
         .withSelector(
-            Collections.singletonMap(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE))
+            Map.of(LABEL_SPARK_CLUSTER_NAME, name, LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE))
         .addNewPort()
         .withName("web")
         .withPort(8081)
@@ -174,7 +175,7 @@ private static StatefulSet buildMasterStatefulSet(
             .withNewMetadataLike(objectMeta)
             .withName(name + "-master")
             .addToLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE)
-            .addToLabels(LABEL_SPARK_VERSION_NAME, version)
+                .addToLabels(LABEL_SPARK_VERSION_NAME, version)
             .withNamespace(namespace)
             .endMetadata()
             .withNewSpecLike(statefulSetSpec)
@@ -186,6 +187,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 +255,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,9 @@ 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 +155,9 @@ 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 +179,9 @@ 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 +246,8 @@ 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