From 554975ded3fce41561e9943b1b3c26410580228e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Tue, 18 Aug 2026 09:39:19 +0200 Subject: [PATCH] fix: retain recently written external resources missing from a stale update An update of the whole resource set of a primary (a poll result or a received event) might have been created before the reconciler wrote a resource, thus not containing it yet. Since such updates are handled as the full actual state, the write was lost from the cache, and the next reconciliation created a duplicate of an already created resource or repeated an already executed update. Writes are now marked as unconfirmed and retained for the next update if it either does not contain the resource at all - the expected case for a create - or still contains a state that a write replaced. Every state replaced since the last update is kept, since the reconciler might write the same resource multiple times in between, and an update created before any of those writes is stale. Any other state is treated as a change made outside of the reconciler and accepted as actual. Marks are dropped on the first update, so a resource really deleted or changed meanwhile is not retained indefinitely. Also guards handleRecentResourceUpdate against a missing cache entry, and resolves the actual resources from the state resources in the external state bulk dependent integration test, which is the recommended approach for resources that take longer to become visible. --- .../dependent-resources.md | 11 ++ .../ExternalResourceCachingEventSource.java | 99 ++++++++++- ...xternalResourceCachingEventSourceTest.java | 162 ++++++++++++++++++ ...ulkDependentResourceExternalWithState.java | 13 +- 4 files changed, 281 insertions(+), 4 deletions(-) diff --git a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md index 8974c41f2a..538c52c00c 100644 --- a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md +++ b/docs/content/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/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java index 109e83b413..e567159a62 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java @@ -67,6 +67,30 @@ public abstract class ExternalResourceCachingEventSource> 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>> unconfirmedWrites = + new ConcurrentHashMap<>(); + + /** + * The last state of a resource written by the reconciler and every state it replaced since the + * last update. There can be multiple replaced states if the reconciler wrote the resource more + * than once without an update in between; an update created before any of those writes is stale. + * The set is empty if the resource was created, since there was no previous state then. + */ + private record RecentWrite(R written, Set replaced) {} + protected ExternalResourceCachingEventSource( Class resourceClass, ResourceIDMapper resourceIDMapper) { this(null, resourceClass, resourceIDMapper); @@ -86,6 +110,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 +130,13 @@ protected synchronized void handleDelete(ResourceID primaryID, Set 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 removedResources = cachedValues == null @@ -131,7 +163,16 @@ protected synchronized void handleResources(ResourceID primaryID, Set newReso protected synchronized void handleResources(Map> 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 +189,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 +198,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 a state that a 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 newResourcesMap) { + var unconfirmed = unconfirmedWrites.remove(primaryID); + if (unconfirmed == null) { + return; + } + unconfirmed.forEach( + (id, write) -> { + var newResource = newResourcesMap.get(id); + if (newResource == null || write.replaced().contains(newResource)) { + 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 cachedResourceMap, Map newResourcesMap) { var addedResources = new HashMap<>(newResourcesMap); @@ -217,6 +287,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso } else { actualValues.computeIfAbsent(resourceId, r -> resource); } + markUnconfirmedWrite(primaryID, resourceId, resource, null); } @Override @@ -226,12 +297,36 @@ 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, resource, previousVersionOfResource); } } } + /** + * Marks the written resource as not confirmed yet by an update, keeping the states replaced by + * previous writes of the same resource. Without those, an update created before an earlier write + * would not be recognized as stale, and the last write would be lost from the cache. + * + * @param replaced the state the write replaced, {@code null} if the resource was created + * @see #unconfirmedWrites + */ + private void markUnconfirmedWrite(ResourceID primaryID, ID resourceId, R written, R replaced) { + unconfirmedWrites + .computeIfAbsent(primaryID, id -> new HashMap<>()) + .compute( + resourceId, + (id, previousWrite) -> { + Set replacedStates = + previousWrite == null ? new HashSet<>() : new HashSet<>(previousWrite.replaced()); + if (replaced != null) { + replacedStates.add(replaced); + } + return new RecentWrite<>(written, replacedStates); + }); + } + @Override public Set getSecondaryResources(P primary) { return getSecondaryResources(ResourceID.fromResource(primary)); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java index 889cc4da75..b200e1e598 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/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,167 @@ 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()); + } + + @Test + void retainsResourceUpdatedTwiceIfUpdateContainsTheStateBeforeBothWrites() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + source.handleRecentResourceUpdate( + primaryID1(), changedTwiceTestResource1(), changedTestResource1()); + + // the update was created before both writes, thus contains the state before the first one + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())) + .containsExactly(changedTwiceTestResource1()); + // no event for the retained resource, only the initial add event + verify(eventHandler, times(1)).handleEvent(new Event(primaryID1())); + } + + @Test + void retainsResourceUpdatedTwiceIfUpdateContainsTheStateBetweenTheWrites() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + source.handleRecentResourceUpdate( + primaryID1(), changedTwiceTestResource1(), changedTestResource1()); + + // the update was created between the two writes, thus contains the intermediate state + source.handleResources(primaryID1(), Set.of(changedTestResource1())); + + assertThat(source.getSecondaryResources(primaryID1())) + .containsExactly(changedTwiceTestResource1()); + verify(eventHandler, times(1)).handleEvent(new Event(primaryID1())); + } + + @Test + void retainsRecentlyCreatedAndThenUpdatedResourceMissingFromUpdate() { + source.handleRecentResourceCreate(primaryID1(), testResource1()); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + + // the update was created before both writes, thus does not contain the resource yet + source.handleResources(primaryID1(), Set.of(testResource2())); + + assertThat(source.getSecondaryResources(primaryID1())) + .containsExactlyInAnyOrder(changedTestResource1(), testResource2()); + } + + @Test + void doesNotRetainResourceUpdatedTwiceIfChangedOutsideOfTheReconciler() { + var externallyChanged = testResource1().setValue("externallyChangedValue"); + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + source.handleRecentResourceUpdate( + primaryID1(), changedTwiceTestResource1(), changedTestResource1()); + + source.handleResources(primaryID1(), Set.of(externallyChanged)); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(externallyChanged); + } + + private static SampleExternalResource changedTestResource1() { + return testResource1().setValue("changedValue"); + } + + private static SampleExternalResource changedTwiceTestResource1() { + return testResource1().setValue("changedValueAgain"); + } + public static class TestExternalCachingEventSource extends ExternalResourceCachingEventSource { public TestExternalCachingEventSource() { diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java index ba08f7fdfa..eb699f225e 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java @@ -135,12 +135,21 @@ public Map 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 getSecondaryResources( ExternalStateBulkDependentCustomResource primary, Context 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