tests

Author: gyfora

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java

@@ -110,15 +110,19 @@ private ObjectMeta createCmObjectMeta(ResourceID uid) {
         var objectMeta = new ObjectMeta();
         objectMeta.setName("autoscaler-" + uid.getName());
         uid.getNamespace().ifPresent(objectMeta::setNamespace);
-        objectMeta.setLabels(
-                Map.of(
-                        Constants.LABEL_COMPONENT_KEY,
-                        LABEL_COMPONENT_AUTOSCALER,
-                        Constants.LABEL_APP_KEY,
-                        uid.getName()));
+        objectMeta.setLabels(getAutoscalerCmLabels(uid));
         return objectMeta;
     }
 
+    @VisibleForTesting
+    public static Map<String, String> getAutoscalerCmLabels(ResourceID uid) {
+        return Map.of(
+                Constants.LABEL_COMPONENT_KEY,
+                LABEL_COMPONENT_AUTOSCALER,
+                Constants.LABEL_APP_KEY,
+                uid.getName());
+    }
+
     private ConfigMap buildConfigMap(HasMetadata cr, ObjectMeta meta) {
         var cm = new ConfigMap();
         cm.setMetadata(meta);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkUtilsTest.java

@@ -27,12 +27,12 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.autoscaler.state.ConfigMapStore;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.utils.Constants;
 import org.apache.flink.kubernetes.utils.KubernetesUtils;
 import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
 
-import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.Container;
 import io.fabric8.kubernetes.api.model.EmptyDirVolumeSource;
@@ -49,6 +49,7 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import org.junit.jupiter.api.Test;
 
 import java.net.HttpURLConnection;
@@ -403,21 +404,39 @@ public void testMergePodUsingArrayName() {
         assertEquals(List.of(v1merged, volume2, volume3), mergedPod.getSpec().getVolumes());
     }
 
+    @Test
+    public void testKubernetesHaMetaDeletion() {
+        var clusterId = "cluster-id";
+        var ns = kubernetesClient.getNamespace();
+        createHAConfigMapWithData("test", ns, clusterId, Map.of("k", "v"));
+
+        // Create autoscaler configmap to test that it's not deleted
+        var autoscalerData =
+                new ConfigMapBuilder()
+                        .withNewMetadata()
+                        .withName("autoscaler-cm")
+                        .withNamespace(ns)
+                        .withLabels(ConfigMapStore.getAutoscalerCmLabels(new ResourceID(clusterId)))
+                        .endMetadata()
+                        .withData(Map.of("a", "s"))
+                        .build();
+        kubernetesClient.resource(autoscalerData).create();
+        FlinkUtils.deleteKubernetesHAMetadata(clusterId, ns, kubernetesClient);
+        assertNotNull(kubernetesClient.resource(autoscalerData).get());
+    }
+
     private void createHAConfigMapWithData(
             String configMapName, String namespace, String clusterId, Map<String, String> data) {
-        final ConfigMap kubernetesConfigMap =
+        var cm =
                 new ConfigMapBuilder()
                         .withNewMetadata()
                         .withName(configMapName)
                         .withNamespace(namespace)
-                        .withLabels(
-                                KubernetesUtils.getConfigMapLabels(
-                                        clusterId,
-                                        Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY))
+                        .withLabels(KubernetesUtils.getCommonLabels(clusterId))
                         .endMetadata()
                         .withData(data)
                         .build();
 
-        kubernetesClient.configMaps().resource(kubernetesConfigMap).createOrReplace();
+        kubernetesClient.configMaps().resource(cm).createOrReplace();
     }
 }