adding service labels to flink config

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java

@@ -377,6 +377,14 @@ public class KubernetesConfigOptions {
                             "The user-specified annotations that are set to the rest Service. The value should be "
                                     + "in the form of a1:v1,a2:v2");
 
+    public static final ConfigOption<Map<String, String>> REST_SERVICE_LABELS =
+            key("kubernetes.rest-service.labels")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The user-specified labels that are set to the rest Service. The value should be "
+                                    + "in the form of a1:v1,a2:v2");
+
     public static final ConfigOption<Map<String, String>> INTERNAL_SERVICE_ANNOTATIONS =
             key("kubernetes.internal-service.annotations")
                     .mapType()
@@ -385,6 +393,14 @@ public class KubernetesConfigOptions {
                             "The user-specified annotations that are set to the internal Service. The value should be "
                                     + "in the form of a1:v1,a2:v2");
 
+    public static final ConfigOption<Map<String, String>> INTERNAL_SERVICE_LABELS =
+            key("kubernetes.internal-service.labels")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The user-specified labels that are set to the internal Service. The value should be "
+                                    + "in the form of a1:v1,a2:v2");
+
     /**
      * Defines the configuration key of that external resource in Kubernetes. This is used as a
      * suffix in an actual config.

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java

@@ -118,12 +118,24 @@ public Map<String, String> getRestServiceAnnotations() {
                 .orElse(Collections.emptyMap());
     }
 
+    public Map<String, String> getRestServiceLabels() {
+        return flinkConfig
+                .getOptional(KubernetesConfigOptions.REST_SERVICE_LABELS)
+                .orElse(Collections.emptyMap());
+    }
+
     public Map<String, String> getInternalServiceAnnotations() {
         return flinkConfig
                 .getOptional(KubernetesConfigOptions.INTERNAL_SERVICE_ANNOTATIONS)
                 .orElse(Collections.emptyMap());
     }
 
+    public Map<String, String> getInternalServiceLabels() {
+        return flinkConfig
+                .getOptional(KubernetesConfigOptions.INTERNAL_SERVICE_LABELS)
+                .orElse(Collections.emptyMap());
+    }
+
     public int getJobManagerMemoryMB() {
         return clusterSpecification.getMasterMemoryMB();
     }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/HeadlessClusterIPService.java

@@ -50,6 +50,7 @@ public Service buildUpInternalService(
                 .withNewMetadata()
                 .withName(serviceName)
                 .withLabels(kubernetesJobManagerParameters.getCommonLabels())
+                .addToLabels(kubernetesJobManagerParameters.getInternalServiceLabels())
                 .withAnnotations(kubernetesJobManagerParameters.getInternalServiceAnnotations())
                 .endMetadata()
                 .withNewSpec()

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ServiceType.java

@@ -53,6 +53,7 @@ public Service buildUpExternalRestService(
                 .withNewMetadata()
                 .withName(serviceName)
                 .withLabels(kubernetesJobManagerParameters.getCommonLabels())
+                .addToLabels(kubernetesJobManagerParameters.getRestServiceLabels())
                 .withAnnotations(kubernetesJobManagerParameters.getRestServiceAnnotations())
                 .endMetadata()
                 .withNewSpec()

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerTestBase.java

@@ -64,6 +64,7 @@ protected void setupFlinkConfig() {
         this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_LABELS, userLabels);
         this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_ANNOTATIONS, userAnnotations);
         this.flinkConfig.set(KubernetesConfigOptions.INTERNAL_SERVICE_ANNOTATIONS, userAnnotations);
+        this.flinkConfig.set(KubernetesConfigOptions.INTERNAL_SERVICE_LABELS, userLabels);
         this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_NODE_SELECTOR, nodeSelector);
         this.flinkConfig.set(
                 JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(JOB_MANAGER_MEMORY));

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecoratorTest.java

@@ -50,12 +50,21 @@ class ExternalServiceDecoratorTest extends KubernetesJobManagerTestBase {
                 }
             };
 
+    private Map<String, String> customizedLabels =
+            new HashMap<String, String>() {
+                {
+                    put("label1", "label-value1");
+                    put("label2", "label-value2");
+                }
+            };
+
     @Override
     protected void onSetup() throws Exception {
         super.onSetup();
 
         this.flinkConfig.set(
                 KubernetesConfigOptions.REST_SERVICE_ANNOTATIONS, customizedAnnotations);
+        this.flinkConfig.set(KubernetesConfigOptions.REST_SERVICE_LABELS, customizedLabels);
         this.externalServiceDecorator =
                 new ExternalServiceDecorator(this.kubernetesJobManagerParameters);
     }
@@ -74,6 +83,7 @@ void testBuildAccompanyingKubernetesResources() throws IOException {
                 .isEqualTo(ExternalServiceDecorator.getExternalServiceName(CLUSTER_ID));
 
         final Map<String, String> expectedLabels = getCommonLabels();
+        expectedLabels.putAll(customizedLabels);
         assertThat(restService.getMetadata().getLabels()).isEqualTo(expectedLabels);
 
         assertThat(restService.getSpec().getType())

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InternalServiceDecoratorTest.java

@@ -67,6 +67,7 @@ void testBuildAccompanyingKubernetesResources() throws IOException {
                 .isEqualTo(InternalServiceDecorator.getInternalServiceName(CLUSTER_ID));
 
         final Map<String, String> expectedLabels = getCommonLabels();
+        expectedLabels.putAll(userLabels);
         assertThat(internalService.getMetadata().getLabels()).isEqualTo(expectedLabels);
 
         assertThat(internalService.getMetadata().getAnnotations()).isEqualTo(userAnnotations);

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java

@@ -109,6 +109,20 @@ void testGetRestServiceAnnotations() {
         assertThat(resultAnnotations).isEqualTo(expectedAnnotations);
     }
 
+    @Test
+    void testGetRestServiceLabels() {
+        final Map<String, String> expectedLabels = new HashMap<>();
+        expectedLabels.put("a1", "v1");
+        expectedLabels.put("a2", "v2");
+
+        flinkConfig.set(KubernetesConfigOptions.REST_SERVICE_LABELS, expectedLabels);
+
+        final Map<String, String> resultLabels =
+                kubernetesJobManagerParameters.getRestServiceLabels();
+
+        assertThat(resultLabels).isEqualTo(expectedLabels);
+    }
+
     @Test
     void testGetInternalServiceAnnotations() {
         final Map<String, String> expectedAnnotations = new HashMap<>();
@@ -123,6 +137,20 @@ void testGetInternalServiceAnnotations() {
         assertThat(resultAnnotations).isEqualTo(expectedAnnotations);
     }
 
+    @Test
+    void testGetInternalServiceLabels() {
+        final Map<String, String> expectedLabels = new HashMap<>();
+        expectedLabels.put("a1", "v1");
+        expectedLabels.put("a2", "v2");
+
+        flinkConfig.set(KubernetesConfigOptions.INTERNAL_SERVICE_LABELS, expectedLabels);
+
+        final Map<String, String> resultLabels =
+                kubernetesJobManagerParameters.getInternalServiceLabels();
+
+        assertThat(resultLabels).isEqualTo(expectedLabels);
+    }
+
     @Test
     void testGetJobManagerMemoryMB() {
         assertThat(kubernetesJobManagerParameters.getJobManagerMemoryMB())