Skip to content

Publish FromConfigMap refreshes as coherent snapshots #4979

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kubernetes-prow merged 3 commits into from
Nov 5, 2026
Merged

Publish FromConfigMap refreshes as coherent snapshots #4979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ad65bb9
Initial plan
brendandburns Nov 3, 2026
b4860cb
Publish FromConfigMap data as coherent live snapshots
brendandburns Nov 3, 2026
dab6bde
Clarify live-view snapshot guarantees and format changes
brendandburns Nov 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 45 additions & 14 deletions ...java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,15 @@
*/
package io.kubernetes.client.spring.extended.manifests;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.kubernetes.client.openapi.models.V1ConfigMap;
import io.kubernetes.client.spring.extended.manifests.annotation.FromConfigMap;
import io.kubernetes.client.spring.extended.manifests.config.KubernetesManifestsProperties;
import io.kubernetes.client.spring.extended.manifests.configmaps.ConfigMapGetter;
import java.lang.reflect.Field;
import java.util.AbstractMap;
import java.util.AbstractSet;
import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
Expand Down Expand Up @@ -77,7 +79,7 @@ public Object postProcessAfterInitialization(Object bean, String beanName) throw
ConfigMapGetter configMapGetter =
getOrCreateConfigMapGetter(fromConfigMapAnnotation, applicationContext);

Cache<String, String> configMapDataCache = Caffeine.newBuilder().build();
ConfigMapData configMapDataCache = new ConfigMapData();
fullyRefreshCache(configMapGetter, fromConfigMapAnnotation, configMapDataCache);
configMapKeyRefresher.scheduleAtFixedRate(
() -> {
Expand All @@ -86,7 +88,7 @@ public Object postProcessAfterInitialization(Object bean, String beanName) throw
manifestsProperties.getRefreshInterval().getSeconds(),
manifestsProperties.getRefreshInterval().getSeconds(),
TimeUnit.SECONDS);
ReflectionUtils.setField(field, bean, configMapDataCache.asMap());
ReflectionUtils.setField(field, bean, Collections.unmodifiableMap(configMapDataCache));
}

return bean;
Expand All @@ -95,19 +97,48 @@ public Object postProcessAfterInitialization(Object bean, String beanName) throw
private static void fullyRefreshCache(
ConfigMapGetter configMapGetter,
FromConfigMap fromConfigMapAnnotation,
Cache<String, String> configMapDataCache) {
ConfigMapData configMapDataCache) {
V1ConfigMap configMap =
configMapGetter.get(fromConfigMapAnnotation.namespace(), fromConfigMapAnnotation.name());
if (configMap == null || configMap.getData() == null) {
configMapDataCache.invalidateAll();
return;
configMapDataCache.snapshot =
configMap == null || configMap.getData() == null
? Map.of()
: Map.copyOf(configMap.getData());
}

private static final class ConfigMapData extends AbstractMap<String, String> {
private volatile Map<String, String> snapshot = Map.of();

@Override
public String get(Object key) {
return snapshot.get(key);
}

@Override
public String getOrDefault(Object key, String defaultValue) {
return snapshot.getOrDefault(key, defaultValue);
}

@Override
public boolean containsKey(Object key) {
return snapshot.containsKey(key);
}

@Override
public Set<Entry<String, String>> entrySet() {
// Keep the unmodifiable wrapper's cached views live; each iterator captures a snapshot.
return new AbstractSet<>() {
@Override
public Iterator<Entry<String, String>> iterator() {
return snapshot.entrySet().iterator();
}

@Override
public int size() {
return snapshot.size();
}
};
}
// TODO: make the cache data refreshment atomic
Map<String, String> newData = configMap.getData();
newData.forEach(configMapDataCache::put);
Set.copyOf(configMapDataCache.asMap().keySet()).stream()
.filter(key -> !newData.containsKey(key))
.forEach(configMapDataCache::invalidate);
}

private ConfigMapGetter getOrCreateConfigMapGetter(
Expand Down
13 changes: 13 additions & 0 deletions ...rc/main/java/io/kubernetes/client/spring/extended/manifests/annotation/FromConfigMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,20 @@
* <p>The content of the map will be automatically updated at the interval specified by the property
* "kubernetes.manifests.refreshInterval".
*
* <p>The injected map reference remains live: each successful refresh atomically publishes a
* complete, immutable copy of the ConfigMap data. Iteration (including {@code forEach}) and copies
* such as {@code new HashMap<>(map)} observe one complete version. Views returned by {@code
* entrySet()}, {@code keySet()}, and {@code values()} remain live, but each iterator retains the
* version captured when it was created. Separate operations on the map or its views (for example,
* two {@code get} calls, or {@code size()} followed by iteration) may observe different versions if
* a refresh occurs between them; they are not a transaction.
*
* <p>Unlike the previously injected mutable cache map, the map and its views are now read-only.
* Mutating operations throw {@link UnsupportedOperationException}. Copy the map into an
* application-owned map if local modifications are needed.
*
* <p>If the given configmap, is not present in the cluster, the content of the map will stay empty.
* A deleted ConfigMap or a ConfigMap with null data publishes an empty snapshot.
*/
@Target({ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
Expand Down
168 changes: 167 additions & 1 deletion ...test/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,26 @@
*/
package io.kubernetes.client.spring.extended.manifests;

import static org.awaitility.Awaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;

import io.kubernetes.client.openapi.models.V1ConfigMap;
import io.kubernetes.client.spring.extended.manifests.annotation.FromConfigMap;
import io.kubernetes.client.spring.extended.manifests.config.KubernetesManifestsProperties;
import io.kubernetes.client.spring.extended.manifests.configmaps.ConfigMapGetter;
import java.time.Duration;
import java.util.AbstractMap;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -109,6 +120,161 @@ void keyUpdate() {
});
}

@Test
void publishesCompleteSnapshotsThroughRetainedReference() throws Exception {
Map<String, String> original = myBean.dynamicData;
Map<String, String> oldData = Map.of("foo", "old", "related", "old", "obsolete", "old");
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap().data(oldData));
await().until(() -> new HashMap<>(original).equals(oldData));

Set<Map.Entry<String, String>> entries = original.entrySet();
Set<String> keys = original.keySet();
var values = original.values();
Iterator<Map.Entry<String, String>> oldEntries = entries.iterator();
Iterator<String> oldKeys = keys.iterator();
Iterator<String> oldValues = values.iterator();
Map<String, String> newData = Map.of("foo", "new", "related", "new", "added", "new");
PausingMap source = new PausingMap(newData);
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap().data(source));
try {
assertThat(source.copyStarted.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(new HashMap<>(original)).isEqualTo(oldData);
assertThat(Map.copyOf(original)).isEqualTo(oldData);
Map<String, String> visited = new HashMap<>();
original.forEach(visited::put);
assertThat(visited).isEqualTo(oldData);
} finally {
source.continueCopy.countDown();
}

await().until(() -> new HashMap<>(original).equals(newData));
assertThat(myBean.dynamicData).isSameAs(original);
Map<String, String> retainedEntries = new HashMap<>();
oldEntries.forEachRemaining(entry -> retainedEntries.put(entry.getKey(), entry.getValue()));
assertThat(retainedEntries).isEqualTo(oldData);
List<String> retainedKeys = new ArrayList<>();
oldKeys.forEachRemaining(retainedKeys::add);
assertThat(retainedKeys).containsExactlyInAnyOrderElementsOf(oldData.keySet());
List<String> retainedValues = new ArrayList<>();
oldValues.forEachRemaining(retainedValues::add);
assertThat(retainedValues).containsExactlyInAnyOrderElementsOf(oldData.values());
assertThat(entries).containsExactlyInAnyOrderElementsOf(newData.entrySet());
assertThat(keys).containsExactlyInAnyOrderElementsOf(newData.keySet());
assertThat(values).containsExactlyInAnyOrderElementsOf(newData.values());

Map<String, String> laterData = Map.of("foo", "later");
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap().data(laterData));
await().until(() -> Map.copyOf(original).equals(laterData));
}

@Test
void forEachRetainsSnapshotWhenRefreshOccursDuringTraversal() {
Map<String, String> oldData = Map.of("foo", "old", "related", "old", "obsolete", "old");
Map<String, String> newData = Map.of("foo", "new", "related", "new");
Map<String, String> original = myBean.dynamicData;
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap().data(oldData));
await().until(() -> Map.copyOf(original).equals(oldData));

Map<String, String> visited = new HashMap<>();
original.forEach(
(key, value) -> {
if (visited.isEmpty()) {
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap().data(newData));
await().until(() -> Map.copyOf(original).equals(newData));
}
visited.put(key, value);
});
assertThat(visited).isEqualTo(oldData);
assertThat(Map.copyOf(original)).isEqualTo(newData);
}

@Test
void deletionAndNullDataPublishEmptySnapshots() {
Map<String, String> original = myBean.dynamicData;
mockAtomicConfigMapGetter.configMapAtomicReference.set(null);
await().until(original::isEmpty);
mockAtomicConfigMapGetter.configMapAtomicReference.set(
new V1ConfigMap().putDataItem("foo", "restored"));
await().until(() -> "restored".equals(original.get("foo")));
mockAtomicConfigMapGetter.configMapAtomicReference.set(new V1ConfigMap());
await().until(original::isEmpty);
assertThat(myBean.dynamicData).isSameAs(original);
}

@Test
void publishedMapAndViewsAreReadOnly() {
Map<String, String> data = myBean.dynamicData;
assertThatThrownBy(() -> data.put("foo", "local"))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.remove("foo")).isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(data::clear).isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.putAll(Map.of("local", "value")))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.compute("foo", (key, value) -> "local"))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.replaceAll((key, value) -> "local"))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.entrySet().iterator().next().setValue("local"))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.keySet().remove("foo"))
.isInstanceOf(UnsupportedOperationException.class);
assertThatThrownBy(() -> data.values().remove("bar1"))
.isInstanceOf(UnsupportedOperationException.class);
Iterator<Map.Entry<String, String>> iterator = data.entrySet().iterator();
iterator.next();
assertThatThrownBy(iterator::remove).isInstanceOf(UnsupportedOperationException.class);
assertThat(data).containsExactlyEntriesOf(Map.of("foo", "bar1"));
}

private static class PausingMap extends AbstractMap<String, String> {
private final Map<String, String> data;
private final CountDownLatch copyStarted = new CountDownLatch(1);
private final CountDownLatch continueCopy = new CountDownLatch(1);

private PausingMap(Map<String, String> data) {
this.data = new LinkedHashMap<>(data);
}

@Override
public Set<Entry<String, String>> entrySet() {
return new AbstractSet<>() {
@Override
public int size() {
return data.size();
}

@Override
public Iterator<Entry<String, String>> iterator() {
Iterator<Entry<String, String>> delegate = data.entrySet().iterator();
return new Iterator<>() {
private int count;

@Override
public boolean hasNext() {
return delegate.hasNext();
}

@Override
public Entry<String, String> next() {
if (++count == 2) {
copyStarted.countDown();
try {
if (!continueCopy.await(5, TimeUnit.SECONDS)) {
throw new AssertionError("Timed out waiting to finish copying ConfigMap data");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError(e);
}
}
return delegate.next();
}
};
}
};
}
}

static class MockConfigMapGetter implements ConfigMapGetter {
@Override
public V1ConfigMap get(String namespace, String name) {
Expand Down
Loading