wip

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AbstractConfigurationService.java

@@ -26,6 +26,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
 import io.javaoperatorsdk.operator.processing.event.source.informer.pool.DefaultInformerPool;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 /**
  * An abstract implementation of {@link ConfigurationService} meant to ease custom implementations
@@ -195,7 +196,7 @@ public ExecutorServiceManager getExecutorServiceManager() {
   }
 
   @Override
-  public synchronized AbstractInformerPool informerPool() {
+  public synchronized InformerPool informerPool() {
     // cached so that all controllers backed by this ConfigurationService share the same pool and
     // can therefore share the underlying informers; synchronized so concurrent first-access from
     // multiple controllers cannot create (and share out) more than one pool instance

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -44,7 +44,6 @@
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflowFactory;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
-import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
 import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 /** An interface from which to retrieve configuration information. */
@@ -492,14 +491,10 @@ default boolean cloneSecondaryResourcesWhenGettingFromCache() {
    * and would hand out a fresh (unshared) pool on each call; {@link AbstractConfigurationService}
    * provides a cached implementation backed by the default sharing pool.
    *
-   * <p>Typed as {@link AbstractInformerPool} rather than as the narrower {@link InformerPool}
-   * contract consumed by the event sources, because the framework also has to inject this
-   * configuration service into the pool.
-   *
    * @return the informer pool for this configuration service
    */
   @Experimental(
       "Only the configuration API around informer pooling could still change in a"
           + " non-backwards-compatible way, the pooling itself is prod ready.")
-  AbstractInformerPool informerPool();
+  InformerPool informerPool();
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java

@@ -31,6 +31,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.Experimental;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
 import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
 
 @SuppressWarnings({"unused", "UnusedReturnValue"})
 public class ConfigurationServiceOverrider {
@@ -55,7 +56,7 @@ public class ConfigurationServiceOverrider {
   private Set<Class<? extends HasMetadata>> defaultNonSSAResource;
   private Boolean useSSAToPatchPrimaryResource;
   private Boolean cloneSecondaryResourcesWhenGettingFromCache;
-  private AbstractInformerPool informerPool;
+  private InformerPool informerPool;
 
   @SuppressWarnings("rawtypes")
   private DependentResourceFactory dependentResourceFactory;
@@ -329,7 +330,7 @@ public boolean cloneSecondaryResourcesWhenGettingFromCache() {
       }
 
       @Override
-      public AbstractInformerPool informerPool() {
+      public InformerPool informerPool() {
         if (informerPool == null) {
           return super.informerPool();
         }

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

@@ -59,7 +59,7 @@ public ConfigurationService getConfigurationService() {
     return configurationService;
   }
 
-  /** Injecting the configuration service, so users have a cleaner API creating the pool. */
+  @Override
   public void setConfigurationService(ConfigurationService configurationService) {
     this.configurationService = configurationService;
   }

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

@@ -78,4 +78,21 @@ <R extends HasMetadata> void start(
    */
   <R extends HasMetadata> Optional<SharedIndexInformer<R>> releaseInformer(
       String controllerName, String name, InformerClassifier<R> classifier);
+
+  /**
+   * Binds this pool to the {@link ConfigurationService} it belongs to. Called by the framework when
+   * the pool is resolved from that configuration service, before the pool is used; users are not
+   * expected to call it themselves.
+   *
+   * <p>The pool needs the configuration service to create and start informers: the {@link
+   * ConfigurationService#cacheSyncTimeout()} to wait for, whether to {@link
+   * ConfigurationService#stopOnInformerErrorDuringStartup()}, and the {@link
+   * ConfigurationService#getInformerStoppedHandler()} to hook up.
+   *
+   * <p>Injecting it here, rather than requiring it as a constructor argument, is what keeps
+   * creating a pool a plain {@code new NonSharingInformerPool()} for users configuring one through
+   * {@link io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider#withInformerPool}.
+   * A pool instance therefore belongs to exactly one configuration service.
+   */
+  void setConfigurationService(ConfigurationService configurationService);
 }

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AbstractSharedInformerIT.java

@@ -24,6 +24,7 @@
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider;
 import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
@@ -81,7 +82,8 @@ void bothControllersReconcileWatchingConfigMap() {
                   .isPositive();
             });
 
-    var pool = extension.getOperator().getConfigurationService().informerPool();
+    var pool =
+        (AbstractInformerPool) extension.getOperator().getConfigurationService().informerPool();
 
     // the ConfigMap informer count depends on the pool strategy (shared vs. one-per-controller)
     assertThat(pool.numberOfInformersForResource(ConfigMap.class))

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AbstractDeregisterSharedInformerIT.java

@@ -23,6 +23,7 @@
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider;
 import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
@@ -54,7 +55,8 @@ public abstract class AbstractDeregisterSharedInformerIT {
   @Test
   void deregisteringDynamicEventSourceRemovesInformerFromPool() {
     var reconciler = extension.getReconcilerOfType(DeregisterReconciler.class);
-    var pool = extension.getOperator().getConfigurationService().informerPool();
+    var pool =
+        (AbstractInformerPool) extension.getOperator().getConfigurationService().informerPool();
 
     // Create the primary with registration enabled: the reconciler dynamically registers the event
     // source for the watched resource.

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AbstractDynamicSharedInformerIT.java

@@ -24,6 +24,7 @@
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider;
 import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
@@ -95,7 +96,8 @@ void dynamicallyRegisteredEventSourceReceivesInitialEvent() {
     await().untilAsserted(() -> assertThat(dynamicReconciler.getNumberOfExecutions()).isPositive());
 
     // (1) Informer count for the third resource, which depends on the pool strategy.
-    var pool = extension.getOperator().getConfigurationService().informerPool();
+    var pool =
+        (AbstractInformerPool) extension.getOperator().getConfigurationService().informerPool();
     await()
         .untilAsserted(
             () ->