This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 59a6ffde [FLINK-40587] Recreating a FlinkDeployment throws an
autoscaler exception
59a6ffde is described below
commit 59a6ffde793c2eaf91fe1c47b0b82c5dc9214f01
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Mon Sep 14 17:33:34 2026 +0200
[FLINK-40587] Recreating a FlinkDeployment throws an autoscaler exception
---
.../operator/autoscaler/state/ConfigMapStore.java | 29 +++-
.../operator/autoscaler/state/ConfigMapView.java | 68 ++++++++-
.../autoscaler/state/ConfigMapStoreTest.java | 82 +++++++++++
.../autoscaler/state/ConfigMapViewTest.java | 156 +++++++++++++++++++++
4 files changed, 324 insertions(+), 11 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java
index 2abb1420..ad41c8b2 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStore.java
@@ -25,12 +25,15 @@ import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.net.HttpURLConnection;
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@@ -75,16 +78,22 @@ public class ConfigMapStore {
public void flush(KubernetesJobAutoScalerContext jobContext) {
ConfigMapView configMapView = cache.get(jobContext.getJobKey());
if (configMapView == null) {
- LOG.debug("The configMap doesn't exist, so skip the flush.");
+ LOG.debug("The autoscaler ConfigMap doesn't exist, so skip the
flush.");
return;
}
try {
configMapView.flush();
} catch (Exception e) {
+ removeInfoFromCache(jobContext.getJobKey());
+ if (isNotFound(e)) {
+ LOG.warn(
+ "Autoscaler ConfigMap of {} no longer exists, it will
be recreated on the next cycle.",
+ jobContext.getJobKey().getName());
+ return;
+ }
LOG.error(
- "Error while updating autoscaler info configmap,
invalidating to clear the cache",
+ "Error while updating autoscaler ConfigMap, invalidating
to clear the cache.",
e);
- removeInfoFromCache(jobContext.getJobKey());
throw e;
}
}
@@ -93,9 +102,19 @@ public class ConfigMapStore {
cache.remove(resourceID);
}
+ private static boolean isNotFound(Exception e) {
+ return e instanceof KubernetesClientException
+ && ((KubernetesClientException) e).getCode() ==
HttpURLConnection.HTTP_NOT_FOUND;
+ }
+
private ConfigMapView getConfigMap(KubernetesJobAutoScalerContext
jobContext) {
- return cache.computeIfAbsent(
- jobContext.getJobKey(), (id) ->
getConfigMapFromKubernetes(jobContext));
+ var ownerUid = jobContext.getResource().getMetadata().getUid();
+ return cache.compute(
+ jobContext.getJobKey(),
+ (id, cached) ->
+ cached != null && Objects.equals(ownerUid,
cached.getOwnerUid())
+ ? cached
+ : getConfigMapFromKubernetes(jobContext));
}
@VisibleForTesting
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java
index 8a6a37c6..88a0eccb 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapView.java
@@ -21,12 +21,15 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.dsl.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.net.HttpURLConnection;
import java.util.Collections;
import java.util.Map;
+import java.util.Objects;
import java.util.function.Function;
class ConfigMapView {
@@ -36,7 +39,10 @@ class ConfigMapView {
enum State {
/** ConfigMap is only stored locally, not created in Kubernetes yet. */
NEEDS_CREATE,
- /** ConfigMap exists in Kubernetes but there are newer local changes.
*/
+ /**
+ * ConfigMap exists in Kubernetes but there are newer local changes.
Falls back to
+ * NEEDS_CREATE if the ConfigMap was already deleted when flushing.
+ */
NEEDS_UPDATE,
/** ConfigMap view reflects the actual contents of Kubernetes
ConfigMap. */
UP_TO_DATE
@@ -46,14 +52,18 @@ class ConfigMapView {
private ConfigMap configMap;
+ private final String ownerUid;
+
private final Function<ConfigMap, Resource<ConfigMap>> resourceRetriever;
public ConfigMapView(
ConfigMap configMapSkeleton,
Function<ConfigMap, Resource<ConfigMap>> resourceRetriever) {
+ this.ownerUid = ownerUidOf(configMapSkeleton);
var existingConfigMap =
resourceRetriever.apply(configMapSkeleton).get();
if (existingConfigMap != null) {
refreshConfigMap(existingConfigMap);
+ takeOverIfLeftByPreviousOwner(configMapSkeleton);
} else {
this.configMap = configMapSkeleton;
this.state = State.NEEDS_CREATE;
@@ -89,13 +99,54 @@ class ConfigMapView {
if (state == State.UP_TO_DATE) {
return;
}
- Resource<ConfigMap> resource = resourceRetriever.apply(configMap);
if (state == State.NEEDS_UPDATE) {
- refreshConfigMap(resource.update());
- } else if (state == State.NEEDS_CREATE) {
- LOG.info("Creating config map {}",
configMap.getMetadata().getName());
- refreshConfigMap(resource.create());
+ try {
+ refreshConfigMap(resourceRetriever.apply(configMap).update());
+ return;
+ } catch (KubernetesClientException e) {
+ if (e.getCode() != HttpURLConnection.HTTP_NOT_FOUND) {
+ throw e;
+ }
+ // The ConfigMap was deleted in the meantime, typically by the
Kubernetes garbage
+ // collector after the owning resource was replaced. So, we
fallback to create it
+ // with the current local state in order to not lose the
current autoscaling
+ // cycle of computations.
+ LOG.info(
+ "ConfigMap {} no longer exists, recreating it",
+ configMap.getMetadata().getName());
+ configMap.getMetadata().setResourceVersion(null);
+ configMap.getMetadata().setUid(null);
+ state = State.NEEDS_CREATE;
+ }
}
+ LOG.info("Creating ConfigMap {}", configMap.getMetadata().getName());
+ refreshConfigMap(resourceRetriever.apply(configMap).create());
+ }
+
+ /**
+ * A ConfigMap left behind by a previous instance of the resource is taken
over rather than left
+ * to the Kubernetes garbage collector, because in this way it may still
reference a stale
+ * object that can generate a 404 error at flushing time. Its contents are
also dropped because
+ * they reference a resource that is no longer existing.
+ */
+ private void takeOverIfLeftByPreviousOwner(ConfigMap configMapSkeleton) {
+ if (Objects.equals(ownerUid, ownerUidOf(configMap))) {
+ return;
+ }
+ configMap
+ .getMetadata()
+
.setOwnerReferences(configMapSkeleton.getMetadata().getOwnerReferences());
+ configMap.getData().clear();
+ requireUpdate();
+ }
+
+ String getOwnerUid() {
+ return ownerUid;
+ }
+
+ private static String ownerUidOf(ConfigMap configMap) {
+ var ownerReferences = configMap.getMetadata().getOwnerReferences();
+ return ownerReferences.isEmpty() ? null :
ownerReferences.get(0).getUid();
}
private void refreshConfigMap(ConfigMap configMap) {
@@ -114,4 +165,9 @@ class ConfigMapView {
public Map<String, String> getDataReadOnly() {
return Collections.unmodifiableMap(configMap.getData());
}
+
+ @VisibleForTesting
+ protected ConfigMap getConfigMap() {
+ return configMap;
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStoreTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStoreTest.java
index 9c6b8fb2..ff7f86c5 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStoreTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapStoreTest.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
import static
org.apache.flink.kubernetes.operator.autoscaler.TestingKubernetesAutoscalerUtils.createContext;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -140,6 +141,87 @@ public class ConfigMapStoreTest {
assertThat(configMapStore.getSerializedState(ctx, "a")).contains("2");
}
+ @Test
+ void testCacheInvalidatedWhenOwnerUidChanges() {
+ var ctx = createContext("cr1", kubernetesClient);
+ ctx.getResource().getMetadata().setUid("uid-1");
+
+ configMapStore.putSerializedState(ctx, "a", "1");
+ configMapStore.flush(ctx);
+ var requestCount = mockWebServer.getRequestCount();
+
+ assertThat(configMapStore.getSerializedState(ctx, "a")).contains("1");
+ assertEquals(requestCount, mockWebServer.getRequestCount());
+
+ ctx.getResource().getMetadata().setUid("uid-2");
+ assertThat(configMapStore.getSerializedState(ctx, "a")).isEmpty();
+ assertEquals(requestCount + 1, mockWebServer.getRequestCount());
+ }
+
+ @Test
+ void testRecreatedResourceTakesOverTheConfigMapLeftBehind() {
+ var ctx = createContext("cr1", kubernetesClient);
+ ctx.getResource().getMetadata().setUid("uid-1");
+ configMapStore.putSerializedState(ctx, "old", "state");
+ configMapStore.flush(ctx);
+
+ ctx.getResource().getMetadata().setUid("uid-2");
+
+ configMapStore.putSerializedState(ctx, "x", "1");
+ configMapStore.putSerializedState(ctx, "y", "2");
+ configMapStore.flush(ctx);
+
+ var stored = configMapStore.getConfigMapFromKubernetes(ctx);
+ assertThat(stored.getDataReadOnly()).containsEntry("x",
"1").containsEntry("y", "2");
+ assertThat(stored.getDataReadOnly()).doesNotContainKey("old");
+ assertThat(stored.getOwnerUid()).isEqualTo("uid-2");
+ }
+
+ @Test
+ void testFlushRecreatesConfigMapDeletedExternally() {
+ var ctx = createContext("cr1", kubernetesClient);
+
+ configMapStore.putSerializedState(ctx, "a", "1");
+ configMapStore.flush(ctx);
+
+ kubernetesClient
+ .configMaps()
+ .inNamespace(ctx.getResource().getMetadata().getNamespace())
+ .withName("autoscaler-cr1")
+ .delete();
+
+ configMapStore.putSerializedState(ctx, "a", "2");
+ assertDoesNotThrow(() -> configMapStore.flush(ctx));
+ assertThat(configMapStore.getCache()).containsKey(ctx.getJobKey());
+
assertThat(configMapStore.getConfigMapFromKubernetes(ctx).getDataReadOnly())
+ .containsEntry("a", "2");
+ }
+
+ @Test
+ void testTakeOverSurvivesTheCollectorWinningTheRace() {
+ var ctx = createContext("cr1", kubernetesClient);
+ ctx.getResource().getMetadata().setUid("uid-1");
+ configMapStore.putSerializedState(ctx, "old", "state");
+ configMapStore.flush(ctx);
+
+ ctx.getResource().getMetadata().setUid("uid-2");
+ configMapStore.putSerializedState(ctx, "x", "1");
+
+ kubernetesClient
+ .configMaps()
+ .inNamespace(ctx.getResource().getMetadata().getNamespace())
+ .withName("autoscaler-cr1")
+ .delete();
+
+ configMapStore.putSerializedState(ctx, "y", "2");
+ assertDoesNotThrow(() -> configMapStore.flush(ctx));
+
+ var stored = configMapStore.getConfigMapFromKubernetes(ctx);
+ assertThat(stored.getDataReadOnly()).containsEntry("x",
"1").containsEntry("y", "2");
+ assertThat(stored.getDataReadOnly()).doesNotContainKey("old");
+ assertThat(stored.getOwnerUid()).isEqualTo("uid-2");
+ }
+
@Test
void testMinimalAmountOfFlushing() {
KubernetesJobAutoScalerContext ctx = createContext("cr1",
kubernetesClient);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapViewTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapViewTest.java
index f33e862f..b2821e2a 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapViewTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/state/ConfigMapViewTest.java
@@ -18,13 +18,27 @@
package org.apache.flink.kubernetes.operator.autoscaler.state;
import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Proxy;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/** Tests for ConfigMapView. */
@EnableKubernetesMockClient(crud = true)
@@ -87,6 +101,148 @@ public class ConfigMapViewTest {
assertThat(mockWebServer.getRequestCount()).isEqualTo(2);
}
+ @Test
+ void testTakesOverTheConfigMapOfAPreviousOwner() {
+ var leftBehind = createConfigMapSkeleton();
+
leftBehind.getMetadata().setOwnerReferences(List.of(ownerReference("uid-1")));
+ leftBehind.setData(Map.of("previous", "state"));
+ kubernetesClient.resource(leftBehind).create();
+
+ var skeleton = createConfigMapSkeleton();
+
skeleton.getMetadata().setOwnerReferences(List.of(ownerReference("uid-2")));
+ var cmView = new ConfigMapView(skeleton, cm ->
kubernetesClient.resource(cm));
+
+ assertThat(cmView.get("previous")).isNull();
+ assertThat(cmView.getOwnerUid()).isEqualTo("uid-2");
+
+ cmView.flush();
+ var persisted =
kubernetesClient.resource(createConfigMapSkeleton()).get();
+
assertThat(persisted.getMetadata().getOwnerReferences().get(0).getUid()).isEqualTo("uid-2");
+ assertThat(persisted.getData().isEmpty()).isTrue();
+ }
+
+ @Test
+ void testFlushRecreatesWhenTheConfigMapVanishedBeforeTheUpdate() {
+ var stored = createConfigMapSkeleton();
+ stored.getMetadata().setResourceVersion("1");
+ stored.getMetadata().setUid("cm-uid-1");
+ stored.setData(new HashMap<>(Map.of("a", "1")));
+
+ var sent = new AtomicReference<ConfigMap>();
+ var updates = new AtomicInteger();
+ var creates = new AtomicInteger();
+ Function<ConfigMap, Resource<ConfigMap>> apiServer =
+ cm -> {
+ sent.set(cm);
+ return mockResource(
+ () -> stored,
+ () -> {
+ updates.incrementAndGet();
+ throw new KubernetesClientException("Not
Found", 404, null);
+ },
+ () -> {
+ creates.incrementAndGet();
+ var meta = sent.get().getMetadata();
+ // A real API server refuses to create an
object that still carries
+ // the identity of the one it replaces. The
mock server does not,
+ // so this is the only place that check is
enforced.
+ if (meta.getResourceVersion() != null ||
meta.getUid() != null) {
+ throw new KubernetesClientException(
+ "resourceVersion should not be set
on objects to be created",
+ 400,
+ null);
+ }
+ return new ConfigMapBuilder(sent.get())
+ .editMetadata()
+ .withResourceVersion("2")
+ .withUid("cm-uid-2")
+ .endMetadata()
+ .build();
+ });
+ };
+
+ var cmView = new ConfigMapView(createConfigMapSkeleton(), apiServer);
+ cmView.put("b", "2");
+
+ cmView.flush();
+ assertThat(updates.get()).isEqualTo(1);
+ assertThat(creates.get()).isEqualTo(1);
+ assertThat(cmView.get("a")).isEqualTo("1");
+ assertThat(cmView.get("b")).isEqualTo("2");
+
assertThat(cmView.getConfigMap().getMetadata().getResourceVersion()).isEqualTo("2");
+
assertThat(cmView.getConfigMap().getMetadata().getUid()).isEqualTo("cm-uid-2");
+
+ // perform one additional flush to check the state remains the same
+ cmView.flush();
+ assertThat(updates.get()).isEqualTo(1);
+ assertThat(creates.get()).isEqualTo(1);
+ assertThat(cmView.get("a")).isEqualTo("1");
+ assertThat(cmView.get("b")).isEqualTo("2");
+
assertThat(cmView.getConfigMap().getMetadata().getResourceVersion()).isEqualTo("2");
+
assertThat(cmView.getConfigMap().getMetadata().getUid()).isEqualTo("cm-uid-2");
+ }
+
+ @Test
+ void testFlushDoesNotRecreateOnOtherUpdateFailures() {
+ var stored = createConfigMapSkeleton();
+ stored.getMetadata().setResourceVersion("1");
+ stored.setData(new HashMap<>(Map.of("a", "1")));
+
+ var creates = new AtomicInteger();
+ Function<ConfigMap, Resource<ConfigMap>> apiServer =
+ cm ->
+ mockResource(
+ () -> stored,
+ () -> {
+ throw new
KubernetesClientException("Conflict", 409, null);
+ },
+ () -> {
+ creates.incrementAndGet();
+ return cm;
+ });
+
+ var cmView = new ConfigMapView(createConfigMapSkeleton(), apiServer);
+ cmView.put("b", "2");
+
+ var failure = assertThrows(KubernetesClientException.class,
cmView::flush);
+ assertThat(failure.getCode()).isEqualTo(409);
+ assertThat(creates.get()).isEqualTo(0);
+ // The pending write is kept for the retry on the next cycle
+ assertThat(cmView.get("b")).isEqualTo("2");
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Resource<ConfigMap> mockResource(
+ Supplier<ConfigMap> get, Supplier<ConfigMap> update,
Supplier<ConfigMap> create) {
+ return (Resource<ConfigMap>)
+ Proxy.newProxyInstance(
+ Resource.class.getClassLoader(),
+ new Class<?>[] {Resource.class},
+ (proxy, method, args) -> {
+ if (args != null && args.length > 0) {
+ throw new
UnsupportedOperationException(method.getName());
+ }
+ return switch (method.getName()) {
+ case "get" -> get.get();
+ case "update" -> update.get();
+ case "create" -> create.get();
+ case "toString" -> "mockResource";
+ case "hashCode" ->
System.identityHashCode(proxy);
+ default -> throw new
UnsupportedOperationException(
+ method.getName());
+ };
+ });
+ }
+
+ private static OwnerReference ownerReference(String uid) {
+ var ownerReference = new OwnerReference();
+ ownerReference.setApiVersion("flink.apache.org/v1beta1");
+ ownerReference.setKind("FlinkDeployment");
+ ownerReference.setName("cr");
+ ownerReference.setUid(uid);
+ return ownerReference;
+ }
+
private boolean keyExistsInKubernetes(String key) {
int requestCount = mockWebServer.getRequestCount();
var cmView2 =
