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() {