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("app", name + "-master", 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("app", name + "-worker", LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE))
         .addNewPort()
         .withName("web")
         .withPort(8081)

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("app", "cluster-name-worker", LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_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("app", "cluster-name-master", LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_MASTER_VALUE),
+        service1.getSpec().getSelector());
   }
 
   @Test