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

[csviri]

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 HashMaps 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.

Part of #3517

[Copilot]

Pull request overview

This PR fixes a concurrency correctness issue in ExternalResourceCachingEventSource by ensuring cache read paths take a synchronized snapshot of per-primary cached resources (the nested maps are HashMaps and are mutated under the event source monitor). This avoids races that could lead to corrupted reads or TOCTOU NullPointerExceptions.

Changes:

• Added a cachedResourcesFor(ResourceID) helper that synchronizes on the event source and returns a snapshot copy of cached secondary resources.
• Routed getSecondaryResources(ResourceID) and the relevant overrides in polling/inbound event sources through the snapshot helper.
• Documented that getCache() returns a live view and that iterating nested maps requires synchronizing on the event source; added a regression test asserting snapshot behavior.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

File	Description
operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java	Introduces synchronized snapshot helper for cache reads and documents thread-safety expectations of getCache().
operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java	Uses cachedResourcesFor to safely snapshot cached resources when registering tasks and when serving getSecondaryResources.
operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java	Uses cachedResourcesFor to safely snapshot cached resources in getSecondaryResources.
operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java	Adds a test verifying getSecondaryResources returns a snapshot, not a live view.

[Copilot]

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (3)

sample-operators/pom.xml:39

• The PR title/description focus on synchronizing reads in ExternalResourceCachingEventSource, but this change also adds a new sample-operators/kotlin-operator module (and updates CI to run it) along with several unrelated framework fixes/version bumps in other files. Consider splitting these concerns into separate PRs or updating the PR title/description to reflect the broader scope.
  operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java:166
• Minor typo: the helper method name acceptedByFiler appears to be meant as acceptedByFilter. Since this code is being touched, consider renaming the method and its call site to avoid perpetuating the typo.

              .anyMatch(r -> acceptedByGenericFilter(r) && acceptedByOnAddFilter(r));

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java:126

• ResourceFetcher.fetchDelay explicitly distinguishes null (no fetch happened yet) from an empty set (fetch happened but no resources were found). cachedResourcesFor(primaryID) returns an empty set for both "no cache entry" and "cached empty", and the current cachedResources.isEmpty() ? null : cachedResources collapses the two cases, potentially changing polling backoff behavior for implementations that rely on the distinction. Consider using cache presence (under the same monitor) to decide whether to pass null vs an empty set.

      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

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (2)

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java:126

• In checkAndRegisterTask, converting cachedResourcesFor(primaryID) to null when isEmpty() loses the distinction between “no cached entry” and “cached but empty”. Previously, an existing empty per-primary cache map resulted in an empty Set (meaning “fetch happened but no resources were found”), but now it becomes null (meaning “no fetch happened”), which can change ResourceFetcher.fetchDelay behavior and scheduling.

    if (scheduledFutures.get(primaryID) == null
        && (registerPredicate == null || registerPredicate.test(resource))) {
      var cachedResources = cachedResourcesFor(primaryID);
      var actualResources = cachedResources.isEmpty() ? null : cachedResources;

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java:223

• The new test getSecondaryResourcesReturnsASnapshotNotALiveView likely passes even without this fix because getSecondaryResources(ResourceID) already returned a copied HashSet before (so it was already a snapshot). To regress the actual TOCTOU/NPE scenario this PR fixes, consider a deterministic test that fails pre-fix by making cache.get(primaryID) return non-null once and null on the second call (simulating a delete between two lookups).

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