Revert "improve:"

This reverts commit faf9bdf.

Author: csviri

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

@@ -65,6 +65,7 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class);
   // we need direct control for the indexer to propagate the just update resource also to the index
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
+  private final TemporalPrimaryToSecondaryIndex<R> temporalPrimaryToSecondaryIndex;
   private final String id = UUID.randomUUID().toString();
 
   public InformerEventSource(
@@ -98,13 +99,17 @@ private InformerEventSource(
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (useSecondaryToPrimaryIndex()) {
+      temporalPrimaryToSecondaryIndex =
+          new DefaultTemporalPrimaryToSecondaryIndex<>(configuration.getSecondaryToPrimaryMapper());
       addIndexers(
           Map.of(
               PRIMARY_TO_SECONDARY_INDEX_NAME,
               (R r) ->
                   configuration.getSecondaryToPrimaryMapper().toPrimaryResourceIDs(r).stream()
                       .map(InformerEventSource::resourceIdToString)
                       .toList()));
+    } else {
+      temporalPrimaryToSecondaryIndex = NOOPTemporalPrimaryToSecondaryIndex.getInstance();
     }
 
     final var informerConfig = configuration.getInformerConfig();
@@ -153,6 +158,7 @@ public void onDelete(R resource, boolean b) {
           ResourceID.fromResource(resource),
           resourceType().getSimpleName());
     }
+    temporalPrimaryToSecondaryIndex.cleanupForResource(resource);
     super.onDelete(resource, b);
     if (acceptedByDeleteFilters(resource, b)) {
       propagateEvent(resource);
@@ -162,6 +168,7 @@ public void onDelete(R resource, boolean b) {
   private synchronized void onAddOrUpdate(
       Operation operation, R newObject, R oldObject, Runnable superOnOp) {
     var resourceID = ResourceID.fromResource(newObject);
+    temporalPrimaryToSecondaryIndex.cleanupForResource(newObject);
     if (canSkipEvent(newObject, oldObject, resourceID)) {
       log.debug(
           "Skipping event propagation for {}, since was a result of a reconcile action. Resource"
@@ -253,10 +260,7 @@ public Set<R> getSecondaryResources(P primary) {
       // that we did not receive yet an event in the informer so the index would not
       // be updated. However, before reading it from temp IDs the event arrives and erases
       // the temp index. So in case of Add not id would be found.
-      var temporalIds =
-          temporaryResourceCache
-              .getTemporalPrimaryToSecondaryIndex()
-              .getSecondaryResources(primaryID);
+      var temporalIds = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID);
       var resources = byIndex(PRIMARY_TO_SECONDARY_INDEX_NAME, resourceIdToString(primaryID));
 
       log.debug(
@@ -309,6 +313,7 @@ public synchronized void handleRecentResourceCreate(ResourceID resourceID, R res
   }
 
   private void handleRecentCreateOrUpdate(R newResource, R oldResource) {
+    temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(newResource);
     temporaryResourceCache.putResource(
         newResource,
         Optional.ofNullable(oldResource)
@@ -365,16 +370,4 @@ private enum Operation {
   private static String resourceIdToString(ResourceID resourceID) {
     return resourceID.getName() + "#" + resourceID.getNamespace().orElse("$na");
   }
-
-  @Override
-  @SuppressWarnings({"unchecked", "rawtypes"})
-  protected TemporaryResourceCache<R> temporaryResourceCache() {
-    return new TemporaryResourceCache<>(
-        this,
-        useSecondaryToPrimaryIndex()
-            ? new DefaultTemporalPrimaryToSecondaryIndex(
-                configuration().getSecondaryToPrimaryMapper())
-            : NOOPTemporalPrimaryToSecondaryIndex.getInstance(),
-        parseResourceVersions);
-  }
 }

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

@@ -40,7 +40,7 @@ public abstract class ManagedInformerEventSource<
 
   private static final Logger log = LoggerFactory.getLogger(ManagedInformerEventSource.class);
   private InformerManager<R, C> cache;
-  protected final boolean parseResourceVersions;
+  private final boolean parseResourceVersions;
   private ControllerConfiguration<R> controllerConfiguration;
   private final C configuration;
   private final Map<String, Function<R, List<String>>> indexers = new HashMap<>();
@@ -87,7 +87,7 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = temporaryResourceCache();
+    temporaryResourceCache = new TemporaryResourceCache<>(this, parseResourceVersions);
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setControllerConfiguration(controllerConfiguration);
     cache.addIndexers(indexers);
@@ -133,11 +133,6 @@ public Optional<R> get(ResourceID resourceID) {
     }
   }
 
-  protected TemporaryResourceCache temporaryResourceCache() {
-    return new TemporaryResourceCache<>(
-        this, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), parseResourceVersions);
-  }
-
   @SuppressWarnings("unused")
   public Optional<R> getCachedValue(ResourceID resourceID) {
     return get(resourceID);

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

@@ -31,6 +31,47 @@
  */
 public class TemporaryResourceCache<T extends HasMetadata> {
 
+  static class ExpirationCache<K> {
+    private final LinkedHashMap<K, Long> cache;
+    private final int ttlMs;
+
+    public ExpirationCache(int maxEntries, int ttlMs) {
+      this.ttlMs = ttlMs;
+      this.cache =
+          new LinkedHashMap<>() {
+            @Override
+            protected boolean removeEldestEntry(Map.Entry<K, Long> eldest) {
+              return size() > maxEntries;
+            }
+          };
+    }
+
+    public void add(K key) {
+      clean();
+      cache.putIfAbsent(key, System.currentTimeMillis());
+    }
+
+    public boolean contains(K key) {
+      clean();
+      return cache.get(key) != null;
+    }
+
+    void clean() {
+      if (!cache.isEmpty()) {
+        long currentTimeMillis = System.currentTimeMillis();
+        var iter = cache.entrySet().iterator();
+        // the order will already be from oldest to newest, clean a fixed number of entries to
+        // amortize the cost amongst multiple calls
+        for (int i = 0; i < 10 && iter.hasNext(); i++) {
+          var entry = iter.next();
+          if (currentTimeMillis - entry.getValue() > ttlMs) {
+            iter.remove();
+          }
+        }
+      }
+    }
+  }
+
   private static final Logger log = LoggerFactory.getLogger(TemporaryResourceCache.class);
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
@@ -40,15 +81,12 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   private final ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
   private final boolean parseResourceVersions;
   private final ExpirationCache<String> knownResourceVersions;
-  private final TemporalPrimaryToSecondaryIndex<T> temporalPrimaryToSecondaryIndex;
 
   public TemporaryResourceCache(
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource,
-      TemporalPrimaryToSecondaryIndex<T> temporalPrimaryToSecondaryIndex,
       boolean parseResourceVersions) {
     this.managedInformerEventSource = managedInformerEventSource;
     this.parseResourceVersions = parseResourceVersions;
-    this.temporalPrimaryToSecondaryIndex = temporalPrimaryToSecondaryIndex;
     if (parseResourceVersions) {
       // keep up to the 50000 add/updates for up to 5 minutes
       knownResourceVersions = new ExpirationCache<>(50000, 600000);
@@ -67,14 +105,10 @@ public synchronized void onAddOrUpdateEvent(T resource) {
   }
 
   synchronized void onEvent(T resource, boolean unknownState) {
-    var res =
-        cache.computeIfPresent(
-            ResourceID.fromResource(resource),
-            (id, cached) ->
-                (unknownState || !isLaterResourceVersion(id, cached, resource)) ? null : cached);
-    if (res == null) {
-      temporalPrimaryToSecondaryIndex.cleanupForResource(resource);
-    }
+    cache.computeIfPresent(
+        ResourceID.fromResource(resource),
+        (id, cached) ->
+            (unknownState || !isLaterResourceVersion(id, cached, resource)) ? null : cached);
   }
 
   public synchronized void putAddedResource(T newResource) {
@@ -120,7 +154,6 @@ public synchronized void putResource(T newResource, String previousResourceVersi
           newResource.getMetadata().getResourceVersion(),
           resourceId);
       cache.put(resourceId, newResource);
-      temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(newResource);
     } else if (cache.remove(resourceId) != null) {
       log.debug("Removed an obsolete resource from cache for id: {}", resourceId);
     }
@@ -156,49 +189,4 @@ private boolean isLaterResourceVersion(ResourceID resourceId, T newResource, T c
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
-
-  public TemporalPrimaryToSecondaryIndex<T> getTemporalPrimaryToSecondaryIndex() {
-    return temporalPrimaryToSecondaryIndex;
-  }
-
-  static class ExpirationCache<K> {
-    private final LinkedHashMap<K, Long> cache;
-    private final int ttlMs;
-
-    public ExpirationCache(int maxEntries, int ttlMs) {
-      this.ttlMs = ttlMs;
-      this.cache =
-          new LinkedHashMap<>() {
-            @Override
-            protected boolean removeEldestEntry(Map.Entry<K, Long> eldest) {
-              return size() > maxEntries;
-            }
-          };
-    }
-
-    public void add(K key) {
-      clean();
-      cache.putIfAbsent(key, System.currentTimeMillis());
-    }
-
-    public boolean contains(K key) {
-      clean();
-      return cache.get(key) != null;
-    }
-
-    void clean() {
-      if (!cache.isEmpty()) {
-        long currentTimeMillis = System.currentTimeMillis();
-        var iter = cache.entrySet().iterator();
-        // the order will already be from oldest to newest, clean a fixed number of entries to
-        // amortize the cost amongst multiple calls
-        for (int i = 0; i < 10 && iter.hasNext(); i++) {
-          var entry = iter.next();
-          if (currentTimeMillis - entry.getValue() > ttlMs) {
-            iter.remove();
-          }
-        }
-      }
-    }
-  }
 }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryPrimaryResourceCacheTest.java

@@ -31,9 +31,7 @@ class TemporaryPrimaryResourceCacheTest {
   @BeforeEach
   void setup() {
     informerEventSource = mock(InformerEventSource.class);
-    temporaryResourceCache =
-        new TemporaryResourceCache<>(
-            informerEventSource, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), false);
+    temporaryResourceCache = new TemporaryResourceCache<>(informerEventSource, false);
   }
 
   @Test
@@ -96,9 +94,7 @@ void removesResourceFromCache() {
 
   @Test
   void resourceVersionParsing() {
-    this.temporaryResourceCache =
-        new TemporaryResourceCache<>(
-            informerEventSource, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), true);
+    this.temporaryResourceCache = new TemporaryResourceCache<>(informerEventSource, true);
 
     assertThat(temporaryResourceCache.isKnownResourceVersion(testResource())).isFalse();