feat: simplify wiring of event sources

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java

@@ -4,20 +4,13 @@
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.Version;
-import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.javaoperatorsdk.operator.api.ResourceController;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.processing.CustomResourceCache;
-import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
-import io.javaoperatorsdk.operator.processing.EventDispatcher;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
-import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
-import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,6 +27,8 @@ public Operator(KubernetesClient k8sClient, ConfigurationService configurationSe
     this.k8sClient = k8sClient;
     this.configurationService = configurationService;
     this.closeables = new ArrayList<>();
+
+    Runtime.getRuntime().addShutdownHook(new Thread(this::close));
   }
 
   /**
@@ -115,18 +110,7 @@ public <R extends CustomResource> void register(
         configuration = existing;
       }
 
-      final var retry = GenericRetry.fromConfiguration(configuration.getRetryConfiguration());
-
-      // check if we only want to watch the current namespace
-      var targetNamespaces = configuration.getNamespaces().toArray(new String[] {});
-      if (configuration.watchCurrentNamespace()) {
-        targetNamespaces =
-            new String[] {configurationService.getClientConfiguration().getNamespace()};
-      }
-
       Class<R> resClass = configuration.getCustomResourceClass();
-      String finalizer = configuration.getFinalizer();
-
       final String controllerName = configuration.getName();
 
       // check that the custom resource is known by the cluster if configured that way
@@ -148,61 +132,18 @@ public <R extends CustomResource> void register(
       }
 
       final var client = k8sClient.customResources(resClass);
-      EventDispatcher<R> dispatcher = new EventDispatcher<>(controller, finalizer, client);
-
-      CustomResourceCache customResourceCache =
-          new CustomResourceCache(configurationService.getObjectMapper());
-      DefaultEventHandler defaultEventHandler =
-          new DefaultEventHandler(
-              customResourceCache,
-              dispatcher,
-              controllerName,
-              retry,
-              configurationService.concurrentReconciliationThreads());
       DefaultEventSourceManager eventSourceManager =
-          new DefaultEventSourceManager(defaultEventHandler, retry != null);
-      defaultEventHandler.setEventSourceManager(eventSourceManager);
-      dispatcher.setEventSourceManager(eventSourceManager);
-
+          new DefaultEventSourceManager(controller, configuration, client);
       controller.init(eventSourceManager);
-      final boolean watchAllNamespaces = configuration.watchAllNamespaces();
-      CustomResourceEventSource customResourceEventSource =
-          createCustomResourceEventSource(
-              client,
-              customResourceCache,
-              watchAllNamespaces,
-              targetNamespaces,
-              configuration.isGenerationAware(),
-              finalizer,
-              resClass);
-
-      closeables.add(customResourceEventSource);
       closeables.add(eventSourceManager);
 
-      customResourceEventSource.setEventHandler(defaultEventHandler);
-      customResourceEventSource.start();
-
       log.info(
           "Registered Controller: '{}' for CRD: '{}' for namespace(s): {}",
           controllerName,
           resClass,
-          watchAllNamespaces ? "[all namespaces]" : Arrays.toString(targetNamespaces));
+          configuration.watchAllNamespaces()
+              ? "[all namespaces]"
+              : configuration.getEffectiveNamespaces());
     }
   }
-
-  private CustomResourceEventSource createCustomResourceEventSource(
-      MixedOperation client,
-      CustomResourceCache customResourceCache,
-      boolean watchAllNamespaces,
-      String[] targetNamespaces,
-      boolean generationAware,
-      String finalizer,
-      Class<?> resClass) {
-
-    return watchAllNamespaces
-        ? CustomResourceEventSource.customResourceEventSourceForAllNamespaces(
-            customResourceCache, client, generationAware, finalizer, resClass)
-        : CustomResourceEventSource.customResourceEventSourceForTargetNamespaces(
-            customResourceCache, client, targetNamespaces, generationAware, finalizer, resClass);
-  }
 }

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

@@ -2,13 +2,10 @@
 
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.processing.event.EventList;
-import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import java.util.Optional;
 
 public interface Context<T extends CustomResource> {
 
-  EventSourceManager getEventSourceManager();
-
   EventList getEvents();
 
   Optional<RetryInfo> getRetryInfo();

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

@@ -2,25 +2,16 @@
 
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.processing.event.EventList;
-import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import java.util.Optional;
 
 public class DefaultContext<T extends CustomResource> implements Context<T> {
 
   private final RetryInfo retryInfo;
   private final EventList events;
-  private final EventSourceManager eventSourceManager;
 
-  public DefaultContext(
-      EventSourceManager eventSourceManager, EventList events, RetryInfo retryInfo) {
+  public DefaultContext(EventList events, RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
     this.events = events;
-    this.eventSourceManager = eventSourceManager;
-  }
-
-  @Override
-  public EventSourceManager getEventSourceManager() {
-    return eventSourceManager;
   }
 
   @Override

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

@@ -35,6 +35,7 @@ private <R extends CustomResource> void put(
       }
     }
     configurations.put(name, config);
+    config.setConfigurationService(this);
   }
 
   protected void throwExceptionOnNameCollision(

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

@@ -41,6 +41,15 @@ static boolean currentNamespaceWatched(Set<String> namespaces) {
         && namespaces.contains(Controller.WATCH_CURRENT_NAMESPACE);
   }
 
+  default Set<String> getEffectiveNamespaces() {
+    var targetNamespaces = getNamespaces();
+    if (watchCurrentNamespace()) {
+      targetNamespaces =
+          Collections.singleton(getConfigurationService().getClientConfiguration().getNamespace());
+    }
+    return targetNamespaces;
+  }
+
   default RetryConfiguration getRetryConfiguration() {
     return RetryConfiguration.DEFAULT;
   }

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

@@ -70,6 +70,11 @@ public ControllerConfiguration<R> build() {
       public Class<R> getCustomResourceClass() {
         return original.getCustomResourceClass();
       }
+
+      @Override
+      public ConfigurationService getConfigurationService() {
+        return original.getConfigurationService();
+      }
     };
   }
 

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

@@ -5,11 +5,14 @@
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.client.dsl.MixedOperation;
+import io.javaoperatorsdk.operator.api.ResourceController;
 import io.javaoperatorsdk.operator.api.RetryInfo;
-import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
+import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 import java.util.HashMap;
@@ -18,6 +21,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,7 +34,6 @@ public class DefaultEventHandler implements EventHandler {
 
   private static final Logger log = LoggerFactory.getLogger(DefaultEventHandler.class);
 
-  private final CustomResourceCache customResourceCache;
   private final EventBuffer eventBuffer;
   private final Set<String> underProcessing = new HashSet<>();
   private final ScheduledThreadPoolExecutor executor;
@@ -43,12 +46,19 @@ public class DefaultEventHandler implements EventHandler {
   private final ReentrantLock lock = new ReentrantLock();
 
   public DefaultEventHandler(
-      CustomResourceCache customResourceCache,
+      ResourceController controller, ControllerConfiguration configuration, MixedOperation client) {
+    this(
+        new EventDispatcher(controller, configuration.getFinalizer(), client),
+        configuration.getName(),
+        GenericRetry.fromConfiguration(configuration.getRetryConfiguration()),
+        configuration.getConfigurationService().concurrentReconciliationThreads());
+  }
+
+  DefaultEventHandler(
       EventDispatcher eventDispatcher,
       String relatedControllerName,
       Retry retry,
       int concurrentReconciliationThreads) {
-    this.customResourceCache = customResourceCache;
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
     this.controllerName = relatedControllerName;
@@ -59,27 +69,14 @@ public DefaultEventHandler(
             runnable -> new Thread(runnable, "EventHandler-" + relatedControllerName));
   }
 
-  public DefaultEventHandler(
-      CustomResourceCache customResourceCache,
-      EventDispatcher eventDispatcher,
-      String relatedControllerName,
-      Retry retry) {
-    this(
-        customResourceCache,
-        eventDispatcher,
-        relatedControllerName,
-        retry,
-        ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER);
-  }
-
   @Override
   public void close() {
-    if (eventSourceManager != null) {
-      log.debug("Closing EventSourceManager {} -> {}", controllerName, eventSourceManager);
-      eventSourceManager.close();
+    try {
+      log.debug("Closing handler for {}", controllerName);
+      executor.awaitTermination(10, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      log.debug("Exception closing handler for {}: {}", controllerName, e.getLocalizedMessage());
     }
-
-    executor.shutdownNow();
   }
 
   public void setEventSourceManager(DefaultEventSourceManager eventSourceManager) {
@@ -102,7 +99,7 @@ private void executeBufferedEvents(String customResourceUid) {
     boolean newEventForResourceId = eventBuffer.containsEvents(customResourceUid);
     boolean controllerUnderExecution = isControllerUnderExecution(customResourceUid);
     Optional<CustomResource> latestCustomResource =
-        customResourceCache.getLatestResource(customResourceUid);
+        eventSourceManager.getLatestResource(customResourceUid);
 
     if (!controllerUnderExecution && newEventForResourceId && latestCustomResource.isPresent()) {
       setUnderExecutionProcessing(customResourceUid);
@@ -234,7 +231,7 @@ private void cacheUpdatedResourceIfChanged(
           getUID(originalCustomResource),
           getVersion(customResourceAfterExecution),
           getVersion(originalCustomResource));
-      this.customResourceCache.cacheResource(
+      eventSourceManager.cacheResource(
           customResourceAfterExecution,
           customResource ->
               getVersion(customResource).equals(originalResourceVersion)
@@ -245,7 +242,6 @@ private void cacheUpdatedResourceIfChanged(
   private void cleanupAfterDeletedEvent(String customResourceUid) {
     eventSourceManager.cleanup(customResourceUid);
     eventBuffer.cleanup(customResourceUid);
-    customResourceCache.cleanup(customResourceUid);
   }
 
   private boolean isControllerUnderExecution(String customResourceUid) {

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

@@ -15,21 +15,19 @@
 import io.javaoperatorsdk.operator.api.ResourceController;
 import io.javaoperatorsdk.operator.api.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.EventList;
-import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Dispatches events to the Controller and handles Finalizers for a single type of Custom Resource.
  */
-public class EventDispatcher<R extends CustomResource> {
+class EventDispatcher<R extends CustomResource> {
 
   private static final Logger log = LoggerFactory.getLogger(EventDispatcher.class);
 
   private final ResourceController<R> controller;
   private final String resourceFinalizer;
   private final CustomResourceFacade<R> customResourceFacade;
-  private EventSourceManager eventSourceManager;
 
   EventDispatcher(
       ResourceController<R> controller,
@@ -47,10 +45,6 @@ public EventDispatcher(
     this(controller, finalizer, new CustomResourceFacade<>(client));
   }
 
-  public void setEventSourceManager(EventSourceManager eventSourceManager) {
-    this.eventSourceManager = eventSourceManager;
-  }
-
   public PostExecutionControl handleExecution(ExecutionScope<R> executionScope) {
     try {
       return handleDispatch(executionScope);
@@ -87,9 +81,7 @@ private PostExecutionControl handleDispatch(ExecutionScope<R> executionScope) {
     }
     Context<R> context =
         new DefaultContext<>(
-            eventSourceManager,
-            new EventList(executionScope.getEvents()),
-            executionScope.getRetryInfo());
+            new EventList(executionScope.getEvents()), executionScope.getRetryInfo());
     if (resource.isMarkedForDeletion()) {
       return handleDelete(resource, context);
     } else {