fix: read the external resource cache under the event source monitor

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java

@@ -238,12 +238,23 @@ public Set<R> getSecondaryResources(P primary) {
   }
 
   public Set<R> 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<R> 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<R> getSecondaryResource(ResourceID primaryID) {
@@ -257,6 +268,12 @@ public Optional<R> 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<ResourceID, Map<ID, R>> getCache() {
     return Collections.unmodifiableMap(cache);
   }

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<R> getAndCacheResource(P primary) {
   @Override
   public Set<R> 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();

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<R> 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();

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<SampleExternalResource, HasMetadata, String> {
     public TestExternalCachingEventSource() {