wip

Signed-off-by: csviri <a_meszaros@apple.com>

Author: csviri

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

@@ -148,7 +148,7 @@ private <R> Void stopEventSource(EventSource<R, P> eventSource) {
     return null;
   }
 
-  @SuppressWarnings("rawtypes")
+  @SuppressWarnings({"rawtypes", "unchecked"})
   public final synchronized <R> void registerEventSource(EventSource<R, P> eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");

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

@@ -99,13 +99,13 @@ private void initSources() {
     final var targetNamespaces =
         configuration.getInformerConfig().getEffectiveNamespaces(controllerConfiguration);
     if (InformerConfiguration.allNamespacesWatched(targetNamespaces)) {
-      var source = createEventSourceForNamespace(WATCH_ALL_NAMESPACES);
+      var source = getAndStartEventSourceForNamespace(WATCH_ALL_NAMESPACES);
       log.debug("Registered {} -> {} for any namespace", this, source);
     } else {
       targetNamespaces.forEach(
           ns -> {
             // todo should we start parallel
-            final var source = createEventSourceForNamespace(ns);
+            final var source = getAndStartEventSourceForNamespace(ns);
 
             log.debug("Registered {} -> {} for namespace: {}", this, source, ns);
           });
@@ -139,18 +139,20 @@ public void changeNamespaces(Set<String> namespaces) {
         .boundedExecuteAndWaitForAllToComplete(
             newNamespaces.stream(),
             ns -> {
-              final var source = createEventSourceForNamespace(ns);
+              final var source = getAndStartEventSourceForNamespace(ns);
               log.debug("Registered new {} -> {} for namespace: {}", this, source, ns);
               return null;
             },
             ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName());
   }
 
-  private InformerWrapper<R> createEventSourceForNamespace(String namespaceIdentifier) {
+  private InformerWrapper<R> getAndStartEventSourceForNamespace(String namespaceIdentifier) {
     final InformerWrapper<R> source;
     InformerClassifier<R> classifier = getClassifier(namespaceIdentifier);
     var informer =
         informerPool.getInformer(configuration.getInformerConfig().getName(), classifier);
+    informerPool.start(informer, classifier);
+
     source =
         new InformerWrapper<>(
             informer, namespaceIdentifier, controllerConfiguration.getConfigurationService());

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

@@ -145,9 +145,24 @@ protected SharedIndexInformer createInformer(InformerClassifier<?> classifier) {
     return informer;
   }
 
-  protected void start(
-      SharedIndexInformer<? extends HasMetadata> informer,
-      InformerClassifier<?> informerClassifier) {
+  /**
+   * Fires (but does not wait for) the informer's start. This is cheap/non-blocking (the actual
+   * list+watch happens asynchronously), so it is safe to call while holding a pool-internal lock.
+   * Doing so atomically with the informer's creation/reference-count bookkeeping is what prevents a
+   * concurrent {@code releaseInformer} from stopping the informer before it is ever started.
+   */
+  protected void triggerStart(
+      SharedIndexInformer<? extends HasMetadata> informer, InformerClassifier<?> classifier) {
+    log.debug(
+        "Starting informer for namespace: {} resource: {}",
+        classifier.namespaceIdentifier(),
+        informer.getApiTypeClass().getSimpleName());
+    informer.start();
+  }
+
+  @Override
+  public <R extends HasMetadata> void start(
+      SharedIndexInformer<R> informer, InformerClassifier<R> informerClassifier) {
     // change thread name for easier debugging
     final var thread = Thread.currentThread();
     final var name = thread.getName();
@@ -159,10 +174,8 @@ protected void start(
               + " "
               + thread.getId());
       final var resourceName = informer.getApiTypeClass().getSimpleName();
-      log.debug(
-          "Starting informer for namespace: {} resource: {}",
-          informerClassifier.namespaceIdentifier(),
-          resourceName);
+      // idempotent: if the informer was already started (e.g. by the pool when it was
+      // created/reused), this just returns the existing start future without restarting it
       var start = informer.start();
       // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is
       // false, and there is a rbac issue the get never returns; therefore operator never really

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

@@ -36,7 +36,6 @@ public class AlwaysCreateInformerPool extends AbstractInformerPool {
   public <R extends HasMetadata> SharedIndexInformer<R> getInformer(
       String name, InformerClassifier<R> classifier) {
     var informer = createInformer(classifier);
-    start(informer, classifier);
     informers.put(new ClassifierWithName(name, classifier), informer);
     return informer;
   }

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

@@ -67,8 +67,12 @@ public <R extends HasMetadata> SharedIndexInformer<R> getInformer(
                         classifier.informerListLimit()));
         counters.get(classifier).incrementAndGet();
       }
+      // fire the start while still holding the lock so a concurrent releaseInformer() can't stop
+      // this informer before it has ever been started; the caller is responsible for actually
+      // waiting on it via start(), which is not done here to avoid blocking the pool lock for the
+      // (potentially slow) cache sync
+      triggerStart(informer, classifier);
     }
-    start(informer, classifier);
     return informer;
   }
 

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

@@ -31,6 +31,15 @@ public interface InformerPool {
   <R extends HasMetadata> SharedIndexInformer<R> getInformer(
       String name, InformerClassifier<R> classifier);
 
+  /**
+   * Starts the informer (if not already started) and blocks until its cache has synced, or the
+   * configured {@link ConfigurationService#cacheSyncTimeout()} elapses. Callers are expected to
+   * invoke this after {@link #getInformer(String, InformerClassifier)} returns; the pool itself
+   * only registers/reference-counts the informer and does not block on cache sync internally.
+   */
+  <R extends HasMetadata> void start(
+      SharedIndexInformer<R> informer, InformerClassifier<R> classifier);
+
   <R extends HasMetadata> Optional<SharedIndexInformer<R>> releaseInformer(
       String name, InformerClassifier<R> classifier);