[FLINK-15153][kubernetes] Service selector needs to contain jobmanage…

…r component label

The jobmanager label needs to be added to service selector. Otherwise, it may select the wrong backend pods(taskmanager).

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ServiceDecorator.java

@@ -33,6 +33,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Setup services port.
@@ -89,7 +90,13 @@ protected Service decorateInternalResource(Service resource, Configuration flink
  }
 
  spec.setPorts(servicePorts);
- spec.setSelector(resource.getMetadata().getLabels());
+
+ final Map<String, String> labels = new LabelBuilder()
+ .withExist(resource.getMetadata().getLabels())
+ .withJobManagerComponent()
+ .toLabels();
+
+ spec.setSelector(labels);
 
  resource.setSpec(spec);
 

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

@@ -106,6 +106,8 @@ public void testCreateInternalService() throws Exception {
 
  assertEquals(KubernetesConfigOptions.ServiceExposedType.ClusterIP.toString(), service.getSpec().getType());
 
+ // The selector labels should contain jobmanager component
+ labels.put(Constants.LABEL_COMPONENT_KEY, Constants.LABEL_COMPONENT_JOB_MANAGER);
  assertEquals(labels, service.getSpec().getSelector());
 
  assertThat(service.getSpec().getPorts().stream().map(ServicePort::getPort).collect(Collectors.toList()),
@@ -133,6 +135,7 @@ public void testCreateRestService() throws Exception {
 
  assertEquals(KubernetesConfigOptions.ServiceExposedType.LoadBalancer.toString(), service.getSpec().getType());
 
+ labels.put(Constants.LABEL_COMPONENT_KEY, Constants.LABEL_COMPONENT_JOB_MANAGER);
  assertEquals(labels, service.getSpec().getSelector());
 
  assertThat(service.getSpec().getPorts().stream().map(ServicePort::getPort).collect(Collectors.toList()),