diff --git a/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapProcessor.java b/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapProcessor.java
index 5312cd04d4..8f22a1045b 100644
--- a/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapProcessor.java
+++ b/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapProcessor.java
@@ -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;
@@ -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(
           () -> {
@@ -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;
@@ -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(
diff --git a/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/annotation/FromConfigMap.java b/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/annotation/FromConfigMap.java
index 240bee9aa1..35b1531f4f 100644
--- a/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/annotation/FromConfigMap.java
+++ b/spring/src/main/java/io/kubernetes/client/spring/extended/manifests/annotation/FromConfigMap.java
@@ -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)
diff --git a/spring/src/test/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapTest.java b/spring/src/test/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapTest.java
index ec3c6bd770..1ce5a348f7 100644
--- a/spring/src/test/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapTest.java
+++ b/spring/src/test/java/io/kubernetes/client/spring/extended/manifests/KubernetesFromConfigMapTest.java
@@ -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;
@@ -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) {