operator-framework · csviri · Aug 18, 2026
diff --git a/...t/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md b/...t/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md
@@ -480,6 +480,17 @@ also be created, one per dependent resource.
 See [integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent)
 as a sample.
 
+Note that an external resource and the state resource referencing it cannot be created atomically:
+the external resource has to be created first, since its identifier is what gets stored in the
+state. If the resources are fetched based on the state - which is usually the case, since the
+identifier is only known from the state - a poll happening in between the two steps cannot see the
+new external resource yet. JOSDK keeps such a recently created resource in the cache for the next
+update to avoid creating a duplicate of it, but for a resource that takes longer to become visible,
+it is recommended to resolve the actual resources from the state resources in
+`BulkDependentResource.getSecondaryResources`, as done in the integration test above. The state
+resources are managed by an `InformerEventSource`, thus are always up-to-date regarding the
+operator's own changes.
+
 ## GenericKubernetesResource based Dependent Resources
 
 In rare circumstances resource handling where there is no class representation or just typeless handling might be

diff --git a/.../javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java b/.../javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java
@@ -67,6 +67,28 @@ public abstract class ExternalResourceCachingEventSource<R, P extends HasMetadat
 
   protected Map<ResourceID, Map<ID, R>> cache = new ConcurrentHashMap<>();
 
+  /**
+   * The resources written by the reconciler ({@link #handleRecentResourceCreate(ResourceID,
+   * Object)} and {@link #handleRecentResourceUpdate(ResourceID, Object, Object)}) that were not
+   * seen yet in a subsequent update of the whole resource set of a primary. Such an update might
+   * have been created (polled or received) before the resource was actually written, thus not
+   * containing the new state yet. Since these updates are handled as the full actual state, the
+   * write would be lost from the cache; the next reconciliation would then create a duplicate of an
+   * already created resource, or repeat an already executed update. Note that a mark is dropped on
+   * the first update, so a resource really deleted or changed in the meantime is not retained
+   * indefinitely.
+   *
+   * @see #retainUnconfirmedWrites(ResourceID, Map)
+   */
+  private final Map<ResourceID, Map<ID, RecentWrite<R>>> unconfirmedWrites =
+      new ConcurrentHashMap<>();
+
+  /**
+   * A resource written by the reconciler and the state it replaced, which is {@code null} in case
+   * the resource was created.
+   */
+  private record RecentWrite<R>(R written, R replaced) {}
+
   protected ExternalResourceCachingEventSource(
       Class<R> resourceClass, ResourceIDMapper<R, ID> resourceIDMapper) {
     this(null, resourceClass, resourceIDMapper);
@@ -86,6 +108,7 @@ protected ExternalResourceCachingEventSource(
   }
 
   protected synchronized void handleDelete(ResourceID primaryID) {
+    unconfirmedWrites.remove(primaryID);
     var res = cache.remove(primaryID);
     if (res != null && deleteAcceptedByFilter(res.values())) {
       getEventHandler().handleEvent(new Event(primaryID));
@@ -105,6 +128,13 @@ protected synchronized void handleDelete(ResourceID primaryID, Set<ID> resourceI
     if (!isRunning()) {
       return;
     }
+    var unconfirmed = unconfirmedWrites.get(primaryID);
+    if (unconfirmed != null) {
+      unconfirmed.keySet().removeAll(resourceIDs);
+      if (unconfirmed.isEmpty()) {
+        unconfirmedWrites.remove(primaryID);
+      }
+    }
     var cachedValues = cache.get(primaryID);
     List<R> removedResources =
         cachedValues == null
@@ -131,7 +161,16 @@ protected synchronized void handleResources(ResourceID primaryID, Set<R> newReso
 
   protected synchronized void handleResources(Map<ResourceID, Set<R>> allNewResources) {
     var toDelete = cache.keySet().stream().filter(k -> !allNewResources.containsKey(k)).toList();
-    toDelete.forEach(this::handleDelete);
+    toDelete.forEach(
+        primaryID -> {
+          if (unconfirmedWrites.containsKey(primaryID)) {
+            // handled as an empty update, so that a recently written resource, that this update
+            // could not see yet, is not removed from the cache
+            handleResources(primaryID, Collections.emptySet());
+          } else {
+            handleDelete(primaryID);
+          }
+        });
     allNewResources.forEach(this::handleResources);
   }
 
@@ -148,6 +187,7 @@ protected synchronized void handleResources(
     }
     var newResourcesMap =
         newResources.stream().collect(Collectors.toMap(resourceIDMapper::idFor, r -> r));
+    retainUnconfirmedWrites(primaryID, newResourcesMap);
     cache.put(primaryID, newResourcesMap);
     if (propagateEvent
         && !newResourcesMap.equals(cachedResources)
@@ -156,6 +196,34 @@ && acceptedByFiler(cachedResources, newResourcesMap)) {
     }
   }
 
+  /**
+   * Keeps the resources written since the received update was created, thus missing from it. An
+   * update is considered stale for a written resource if it does not contain it at all - which is
+   * the expected case for a create - or if it still contains the state that the write replaced. Any
+   * other state is a change that happened outside of the reconciler, so it is accepted as the
+   * actual state.
+   *
+   * @see #unconfirmedWrites
+   */
+  private void retainUnconfirmedWrites(ResourceID primaryID, Map<ID, R> newResourcesMap) {
+    var unconfirmed = unconfirmedWrites.remove(primaryID);
+    if (unconfirmed == null) {
+      return;
+    }
+    unconfirmed.forEach(
+        (id, write) -> {
+          var newResource = newResourcesMap.get(id);
+          if (newResource == null || newResource.equals(write.replaced())) {
+            log.debug(
+                "Retaining recently written resource missing from the update. Primary ID: {},"
+                    + " resource ID: {}",
+                primaryID,
+                id);
+            newResourcesMap.put(id, write.written());
+          }
+        });
+  }
+
   private boolean acceptedByFiler(Map<ID, R> cachedResourceMap, Map<ID, R> newResourcesMap) {
 
     var addedResources = new HashMap<>(newResourcesMap);
@@ -217,6 +285,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso
     } else {
       actualValues.computeIfAbsent(resourceId, r -> resource);
     }
+    markUnconfirmedWrite(primaryID, resourceId, new RecentWrite<>(resource, null));
   }
 
   @Override
@@ -226,12 +295,18 @@ public synchronized void handleRecentResourceUpdate(
     if (actualValues != null) {
       var resourceId = resourceIDMapper.idFor(resource);
       R actualResource = actualValues.get(resourceId);
-      if (actualResource.equals(previousVersionOfResource)) {
+      if (actualResource != null && actualResource.equals(previousVersionOfResource)) {
         actualValues.put(resourceId, resource);
+        markUnconfirmedWrite(
+            primaryID, resourceId, new RecentWrite<>(resource, previousVersionOfResource));
       }
     }
   }
 
+  private void markUnconfirmedWrite(ResourceID primaryID, ID resourceId, RecentWrite<R> write) {
+    unconfirmedWrites.computeIfAbsent(primaryID, id -> new HashMap<>()).put(resourceId, write);
+  }
+
   @Override
   public Set<R> getSecondaryResources(P primary) {
     return getSecondaryResources(ResourceID.fromResource(primary));

diff --git a/...aoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java b/...aoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java
@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source;
 
+import java.util.Map;
 import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -211,6 +212,107 @@ void genericFilteringEvents() {
     verify(eventHandler, times(0)).handleEvent(any());
   }
 
+  @Test
+  void retainsRecentlyCreatedResourceMissingFromUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+
+    // the update was created before the resource, thus does not contain it yet
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1()))
+        .containsExactlyInAnyOrder(testResource1(), testResource2());
+    // no event for the retained resource, only the initial add event
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsRecentlyCreatedResourceOnlyForASingleUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    // this update is created after the resource, so it is really deleted meanwhile
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+    verify(eventHandler, times(2)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void doesNotRetainRecentlyCreatedResourceDeletedBeforeTheUpdate() {
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+    source.handleDelete(primaryID1(), testResource2());
+
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+  }
+
+  @Test
+  void retainsRecentlyCreatedResourceMissingFromWholeCacheUpdate() {
+    source.handleRecentResourceCreate(primaryID1(), testResource1());
+
+    source.handleResources(Map.of());
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+
+    source.handleResources(Map.of());
+
+    assertThat(source.getSecondaryResources(primaryID1())).isEmpty();
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceMissingFromUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    // the update was created before the resource was updated, thus still contains the old state
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1());
+    // no event for the retained resource, only the initial add event
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceOnlyForASingleUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    // this update is created after the resource was updated, so it was really changed meanwhile
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+    verify(eventHandler, times(2)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void doesNotRetainRecentlyUpdatedResourceChangedOutsideOfTheReconciler() {
+    var externallyChanged = testResource1().setValue("externallyChangedValue");
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    source.handleResources(primaryID1(), Set.of(externallyChanged));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(externallyChanged);
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceInWholeCacheUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    source.handleResources(Map.of(primaryID1(), Set.of(testResource1())));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1());
+  }
+
+  private static SampleExternalResource changedTestResource1() {
+    return testResource1().setValue("changedValue");
+  }
+
   public static class TestExternalCachingEventSource
       extends ExternalResourceCachingEventSource<SampleExternalResource, HasMetadata, String> {
     public TestExternalCachingEventSource() {

diff --git a/...dent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java b/...dent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java
@@ -135,12 +135,21 @@ public Map<String, ExternalResource> desiredResources(
     return res;
   }
 
+  /**
+   * Resolves the actual resources from the persisted state instead of the polled cache. An external
+   * resource and the state referencing it cannot be created atomically, so a poll happening in
+   * between replaces the cached resources with the ones it can already see, dropping the freshly
+   * created one. The next reconciliation would then create a duplicate external resource that no
+   * state references anymore, thus is leaked. The state itself is read-after-write consistent,
+   * since it is managed through an {@link
+   * io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource}.
+   */
   @Override
   public Map<String, ExternalResource> getSecondaryResources(
       ExternalStateBulkDependentCustomResource primary,
       Context<ExternalStateBulkDependentCustomResource> context) {
-    var resources = context.getSecondaryResources(ExternalResource.class);
-    return resources.stream().collect(Collectors.toMap(this::externalResourceIndex, r -> r));
+    return fetchResources(primary).stream()
+        .collect(Collectors.toMap(this::externalResourceIndex, r -> r));
   }
 
   @Override