From 9ae684ee3733fc3d215b431ae1fc968c14c82c04 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Wed, 29 Jul 2026 17:58:44 +0200 Subject: [PATCH] fix: read the external resource cache under the event source monitor `ExternalResourceCachingEventSource` mutates its cache from `synchronized` methods (`handleResources`, `handleDelete`, `handleRecentResourceCreate/Update`), but the read paths were not synchronized. The outer map is a `ConcurrentHashMap`; the nested per-primary maps are plain `HashMap`s that `handleDelete` mutates in place, so a reconciler thread reading them while a poll or informer thread writes can observe a corrupted map or throw. `getSecondaryResources(ResourceID)` additionally looked the primary up twice: var cachedValues = cache.get(primaryID); if (cachedValues == null) { return Collections.emptySet(); } else { return new HashSet<>(cache.get(primaryID).values()); } If a concurrent `handleDelete` removes the entry between the two calls, the second `get` returns null and this throws a NullPointerException. Adds a `cachedResourcesFor` helper that snapshots the cached resources while holding the monitor, and routes `getSecondaryResources` plus the `PerResourcePollingEventSource` and `CachingInboundEventSource` overrides (and `checkAndRegisterTask`) through it. The helper only copies, so the potentially slow `ResourceFetcher` calls in those overrides still run outside the lock and cannot block the informer or poll threads. `getCache()` still returns a live view for backwards compatibility, but now documents that iterating the nested maps requires synchronizing on the event source. Adds a test asserting `getSecondaryResources` returns a snapshot rather than a live view. --- .../ExternalResourceCachingEventSource.java | 21 +++++++++++++++++-- .../inbound/CachingInboundEventSource.java | 7 +++---- .../PerResourcePollingEventSource.java | 12 +++++------ ...xternalResourceCachingEventSourceTest.java | 11 ++++++++++ 4 files changed, 38 insertions(+), 13 deletions(-) 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 8a4c476443..78a5a0a3de 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 @@ -238,12 +238,23 @@ public Set getSecondaryResources(P primary) { } public Set getSecondaryResources(ResourceID primaryID) { + return cachedResourcesFor(primaryID); + } + + /** + * Snapshot of the currently cached secondary resources for the target primary. The per-primary + * maps held in the cache are not thread safe and are mutated while holding this object's monitor, + * so they must not be read outside of it. + * + * @param primaryID id of the primary resource + * @return a copy of the cached secondary resources, empty if none are cached + */ + protected synchronized Set cachedResourcesFor(ResourceID primaryID) { var cachedValues = cache.get(primaryID); if (cachedValues == null) { return Collections.emptySet(); - } else { - return new HashSet<>(cache.get(primaryID).values()); } + return new HashSet<>(cachedValues.values()); } public Optional getSecondaryResource(ResourceID primaryID) { @@ -257,6 +268,12 @@ public Optional getSecondaryResource(ResourceID primaryID) { } } + /** + * @return a live, unmodifiable view of the cache. Note that the nested per-primary maps are not + * thread safe and are mutated while holding this object's monitor, so iterating them without + * synchronizing on this event source is not safe. Prefer {@link + * #getSecondaryResources(ResourceID)}, which returns a snapshot. + */ public Map> getCache() { return Collections.unmodifiableMap(cache); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java index 44e0a684b6..8e7518dcb3 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java @@ -16,7 +16,6 @@ package io.javaoperatorsdk.operator.processing.event.source.inbound; import java.util.Collections; -import java.util.HashSet; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -76,9 +75,9 @@ private Set getAndCacheResource(P primary) { @Override public Set getSecondaryResources(P primary) { var primaryID = ResourceID.fromResource(primary); - var cachedValue = cache.get(primaryID); - if (cachedValue != null && !cachedValue.isEmpty()) { - return new HashSet<>(cachedValue.values()); + var cachedValues = cachedResourcesFor(primaryID); + if (!cachedValues.isEmpty()) { + return cachedValues; } else { if (fetchedForPrimaries.contains(primaryID)) { return Collections.emptySet(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java index 0f0eb78a69..886c0ecb05 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java @@ -17,7 +17,6 @@ import java.time.Duration; import java.util.Collections; -import java.util.HashSet; import java.util.Map; import java.util.Optional; import java.util.Set; @@ -123,9 +122,8 @@ private void checkAndRegisterTask(P resource) { var primaryID = ResourceID.fromResource(resource); if (scheduledFutures.get(primaryID) == null && (registerPredicate == null || registerPredicate.test(resource))) { - var cachedResources = cache.get(primaryID); - var actualResources = - cachedResources == null ? null : new HashSet<>(cachedResources.values()); + var cachedResources = cachedResourcesFor(primaryID); + var actualResources = cachedResources.isEmpty() ? null : cachedResources; // note that there is a delay, to not do two fetches when the resources first appeared // and getSecondaryResource is called on reconciliation. scheduleNextExecution(resource, actualResources); @@ -167,9 +165,9 @@ public void run() { @Override public Set getSecondaryResources(P primary) { var primaryID = ResourceID.fromResource(primary); - var cachedValue = cache.get(primaryID); - if (cachedValue != null && !cachedValue.isEmpty()) { - return new HashSet<>(cachedValue.values()); + var cachedValues = cachedResourcesFor(primaryID); + if (!cachedValues.isEmpty()) { + return cachedValues; } else { if (fetchedForPrimaries.contains(primaryID)) { return Collections.emptySet(); 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..4afaccce6d 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 @@ -211,6 +211,17 @@ void genericFilteringEvents() { verify(eventHandler, times(0)).handleEvent(any()); } + @Test + void getSecondaryResourcesReturnsASnapshotNotALiveView() { + source.handleResources(primaryID1(), Set.of(testResource1())); + + var snapshot = source.getSecondaryResources(primaryID1()); + source.handleDelete(primaryID1()); + + assertThat(snapshot).containsExactly(testResource1()); + assertThat(source.getSecondaryResources(primaryID1())).isEmpty(); + } + public static class TestExternalCachingEventSource extends ExternalResourceCachingEventSource { public TestExternalCachingEventSource() {