From 35c64d61bf2104310bd54fa0cc77ee1b641b2c33 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 the state that the write replaced. 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 | 79 +++++++++++++- ...xternalResourceCachingEventSourceTest.java | 102 ++++++++++++++++++ ...ulkDependentResourceExternalWithState.java | 13 ++- 4 files changed, 201 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..d7a50620e5 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,28 @@ 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<>(); + + /** + * 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 written, R replaced) {} + protected ExternalResourceCachingEventSource( Class resourceClass, ResourceIDMapper 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 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 +161,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 +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 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 cachedResourceMap, Map 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 write) { + unconfirmedWrites.computeIfAbsent(primaryID, id -> new HashMap<>()).put(resourceId, write); + } + @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..f106257197 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,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 { 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