[FLINK-36257][K8S] Remove legacy k8s labels

Author: reswqa

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java

@@ -43,8 +43,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
-
 /**
  * Represent {@link KubernetesLeaderElector} in kubernetes. {@link LeaderElector#run()} is a
  * blocking call. It should be run in the IO executor, not the main thread. The lifecycle is bound
@@ -104,8 +102,7 @@ public KubernetesLeaderElector(
                                                 // ConfigMaps.
                                                 .withLabels(
                                                         KubernetesUtils.getConfigMapLabels(
-                                                                leaderConfig.getClusterId(),
-                                                                LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY))
+                                                                leaderConfig.getClusterId()))
                                                 .build(),
                                         leaderConfig.getLockIdentity()))
                         .withRenewDeadline(leaderConfig.getRenewDeadline())

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java

@@ -88,7 +88,6 @@
 import static org.apache.flink.kubernetes.utils.Constants.DNS_POLICY_DEFAULT;
 import static org.apache.flink.kubernetes.utils.Constants.DNS_POLICY_HOSTNETWORK;
 import static org.apache.flink.kubernetes.utils.Constants.JOB_GRAPH_STORE_KEY_PREFIX;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
 import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
 import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
 import static org.apache.flink.kubernetes.utils.Constants.SUBMITTED_JOBGRAPH_FILE_PREFIX;
@@ -192,13 +191,10 @@ public static Map<String, String> getCommonLabels(String clusterId) {
      * the resources.
      *
      * @param clusterId cluster id
-     * @param type the config map use case. It could only be {@link
-     *     Constants#LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY} now.
      * @return Return ConfigMap labels.
      */
-    public static Map<String, String> getConfigMapLabels(String clusterId, String type) {
+    public static Map<String, String> getConfigMapLabels(String clusterId) {
         final Map<String, String> labels = new HashMap<>(getCommonLabels(clusterId));
-        labels.put(Constants.LABEL_CONFIGMAP_TYPE_KEY, type);
         return Collections.unmodifiableMap(labels);
     }
 
@@ -561,9 +557,7 @@ public static void createConfigMapIfItDoesNotExist(
                         new ConfigMapBuilder()
                                 .withNewMetadata()
                                 .withName(configMapName)
-                                .withLabels(
-                                        getConfigMapLabels(
-                                                clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY))
+                                .withLabels(getConfigMapLabels(clusterId))
                                 .endMetadata()
                                 .build());
 

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java

@@ -36,8 +36,6 @@
 import java.util.UUID;
 
 import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_KEY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -153,12 +151,7 @@ void testClusterConfigMapLabelsAreSet() throws Exception {
 
                 assertThat(kubeClient.getConfigMap(configMapName))
                         .hasValueSatisfying(
-                                configMap ->
-                                        assertThat(configMap.getLabels())
-                                                .hasSize(3)
-                                                .containsEntry(
-                                                        LABEL_CONFIGMAP_TYPE_KEY,
-                                                        LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
+                                configMap -> assertThat(configMap.getLabels()).hasSize(2));
             } finally {
                 leaderElector.stop();
             }