feat: share event sources between managed dependents by name

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

@@ -39,6 +39,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceReferencer;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext;
 import io.javaoperatorsdk.operator.health.ControllerHealthInfo;
+import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.WorkflowCleanupResult;
 import io.javaoperatorsdk.operator.processing.event.EventProcessor;
@@ -242,9 +243,23 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
         managedWorkflow.getDependentResourcesWithoutActivationCondition();
     final var size = dependentResourcesByName.size();
     if (size > 0) {
-      dependentResourcesByName.forEach(dependentResource -> {
-        Optional<EventSource> eventSource = dependentResource.eventSource(context);
-        eventSource.ifPresent(eventSourceManager::registerEventSource);
+      dependentResourcesByName.forEach(dr -> {
+        Optional<EventSource> eventSource = dr.eventSource(context);
+
+        eventSource.ifPresent(es -> {
+          eventSourceManager.getEventSourceByName(es.name())
+              .ifPresentOrElse((EventSource regES) -> {
+                if (dr instanceof AbstractEventSourceHolderDependentResource esHolder) {
+                  esHolder.setEventSource(regES);
+                } else {
+                  throw new IllegalStateException("EventSource already registered with name: "
+                      + es.name()
+                      + " and dependent resource is not extends AbstractEventSourceHolderDependentResource: "
+                      + dr.getClass().getName());
+                }
+              },
+                  () -> eventSourceManager.registerEventSource(es));
+        });
       });
 
       // resolve event sources referenced by name for dependents that reuse an existing event source
@@ -306,7 +321,7 @@ public KubernetesClient getClient() {
     return kubernetesClient;
   }
 
-  public MixedOperation<P, KubernetesResourceList<P>, Resource<P>> getCRClient() {
+  public MixedOperation<P, KubernetesResourceList<P>, Resource<P>> getPrimaryResourceClient() {
     return kubernetesClient.resources(configuration.getResourceClass());
   }
 

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

@@ -213,9 +213,9 @@ public <R> List<EventSource<R, P>> getEventSourcesFor(Class<R> dependentType) {
   @Override
   public <R> EventSource<R, P> dynamicallyRegisterEventSource(EventSource<R, P> eventSource) {
     synchronized (this) {
-      var actual = eventSources.existingEventSourceByName(eventSource.name());
+      var actual = eventSources.getEventSourceByName(eventSource.name());
       if (actual != null) {
-        eventSource = actual;
+        eventSource = (EventSource<R, P>) actual;
       } else {
         registerEventSource(eventSource);
       }
@@ -249,6 +249,11 @@ public <R> EventSource<R, P> getEventSourceFor(
     return eventSources.get(dependentType, name);
   }
 
+
+  public Optional<EventSource<?, P>> getEventSourceByName(String name) {
+    return Optional.ofNullable(eventSources.getEventSourceByName(name));
+  }
+
   TimerEventSource<P> retryEventSource() {
     return eventSources.retryEventSource();
   }

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

@@ -19,13 +19,13 @@ class EventSources<P extends HasMetadata> {
 
   private final ConcurrentNavigableMap<String, Map<String, EventSource<?, P>>> sources =
       new ConcurrentSkipListMap<>();
-  private final Map<String, EventSource> sourceByName = new HashMap<>();
+  private final Map<String, EventSource<?, P>> sourceByName = new HashMap<>();
 
   private final TimerEventSource<P> retryAndRescheduleTimerEventSource =
       new TimerEventSource<>("RetryAndRescheduleTimerEventSource");
   private ControllerEventSource<P> controllerEventSource;
 
-  public void add(EventSource eventSource) {
+  public void add(EventSource<?, P> eventSource) {
     final var name = eventSource.name();
     var existing = sourceByName.get(name);
     if (existing != null) {
@@ -47,7 +47,7 @@ public void clear() {
     sourceByName.clear();
   }
 
-  public EventSource existingEventSourceByName(String name) {
+  public EventSource<?, P> getEventSourceByName(String name) {
     return sourceByName.get(name);
   }
 
@@ -92,15 +92,15 @@ private String keyFor(Class<?> dependentType) {
   }
 
   @SuppressWarnings("unchecked")
-  public <S> EventSource<S, P> get(Class<S> dependentType, String name) {
-    if (dependentType == null) {
+  public <S> EventSource<S, P> get(Class<S> resourceType, String name) {
+    if (resourceType == null) {
       throw new IllegalArgumentException("Must pass a dependent type to retrieve event sources");
     }
 
-    final var sourcesForType = sources.get(keyFor(dependentType));
+    final var sourcesForType = sources.get(keyFor(resourceType));
     if (sourcesForType == null || sourcesForType.isEmpty()) {
       throw new IllegalArgumentException(
-          "There is no event source found for class:" + dependentType.getName());
+          "There is no event source found for class:" + resourceType.getName());
     }
 
     final var size = sourcesForType.size();
@@ -110,24 +110,24 @@ public <S> EventSource<S, P> get(Class<S> dependentType, String name) {
     } else {
       if (name == null || name.isBlank()) {
         throw new IllegalArgumentException("There are multiple EventSources registered for type "
-            + dependentType.getCanonicalName()
+            + resourceType.getCanonicalName()
             + ", you need to provide a name to specify which EventSource you want to query. Known names: "
             + String.join(",", sourcesForType.keySet()));
       }
       source = (EventSource<S, P>) sourcesForType.get(name);
 
       if (source == null) {
         throw new IllegalArgumentException("There is no event source found for class:" +
-            " " + dependentType.getName() + ", name:" + name);
+            " " + resourceType.getName() + ", name:" + name);
       }
     }
 
     final var resourceClass = source.resourceType();
-    if (!resourceClass.isAssignableFrom(dependentType)) {
+    if (!resourceClass.isAssignableFrom(resourceType)) {
       throw new IllegalArgumentException(source + " associated with "
-          + keyAsString(dependentType, name)
+          + keyAsString(resourceType, name)
           + " is handling " + resourceClass.getName() + " resources but asked for "
-          + dependentType.getName());
+          + resourceType.getName());
     }
     return source;
   }

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

@@ -58,7 +58,8 @@ class ReconciliationDispatcher<P extends HasMetadata> {
 
   public ReconciliationDispatcher(Controller<P> controller) {
     this(controller,
-        new CustomResourceFacade<>(controller.getCRClient(), controller.getConfiguration()));
+        new CustomResourceFacade<>(controller.getPrimaryResourceClient(),
+            controller.getConfiguration()));
   }
 
   public PostExecutionControl<P> handleExecution(ExecutionScope<P> executionScope) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java

@@ -32,7 +32,7 @@ public class ControllerEventSource<T extends HasMetadata>
 
   @SuppressWarnings({"unchecked", "rawtypes"})
   public ControllerEventSource(Controller<T> controller) {
-    super(NAME, controller.getCRClient(), controller.getConfiguration(), false);
+    super(NAME, controller.getPrimaryResourceClient(), controller.getConfiguration(), false);
     this.controller = controller;
 
     final var config = controller.getConfiguration();

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java

@@ -36,12 +36,6 @@ protected ConfigMap desired(CreateUpdateEventFilterTestCustomResource primary,
         Context<CreateUpdateEventFilterTestCustomResource> context) {
       return desired;
     }
-
-    @Override
-    public void setEventSource(
-        io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource<ConfigMap, CreateUpdateEventFilterTestCustomResource> eventSource) {
-      super.setEventSource(eventSource);
-    }
   }
 
   public static final String CONFIG_MAP_TEST_DATA_KEY = "key";