refactor: EventSourceAware -> DeferrableEventSourceHolder, handle errors

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DeferrableEventSourceHolder.java

@@ -0,0 +1,14 @@
+package io.javaoperatorsdk.operator.api.reconciler.dependent;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+
+public interface DeferrableEventSourceHolder<P extends HasMetadata> {
+
+  default void useEventSourceWithName(String name) {}
+
+  Optional<String> resolveEventSource(EventSourceRetriever<P> eventSourceRetriever);
+
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java

@@ -49,8 +49,6 @@ default Optional<ResourceEventSource<R, P>> eventSource(
     return Optional.empty();
   }
 
-  default void useEventSourceWithName(String name) {}
-
   default Optional<R> getSecondaryResource(P primary, Context<P> context) {
     return Optional.empty();
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/EventSourceAware.java

@@ -1,5 +1,8 @@
 package io.javaoperatorsdk.operator.processing;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -32,7 +35,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceAware;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DeferrableEventSourceHolder;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflow;
@@ -214,25 +217,35 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
       final var ownSources = provider.prepareEventSources(context);
       ownSources.forEach(eventSourceManager::registerEventSource);
     }
-    managedWorkflow
-        .getDependentResourcesByName().entrySet().stream()
-        .forEach(drEntry -> {
-          if (drEntry.getValue() instanceof EventSourceProvider) {
-            final var provider = (EventSourceProvider) drEntry.getValue();
-            final var source = provider.initEventSource(context);
-            eventSourceManager.registerEventSource(drEntry.getKey(), source);
-          } else {
-            Optional<ResourceEventSource> eventSource =
-                drEntry.getValue().eventSource(context);
-            eventSource.ifPresent(es -> {
-              eventSourceManager.registerEventSource(drEntry.getKey(), es);
-            });
-          }
-        });
-    managedWorkflow.getDependentResourcesByName().entrySet().stream().map(Map.Entry::getValue)
-        .filter(EventSourceAware.class::isInstance)
-        .forEach(dr -> ((EventSourceAware) dr)
-            .selectEventSources(eventSourceManager));
+
+    // register created event sources
+    final var dependentResourcesByName = managedWorkflow.getDependentResourcesByName();
+    final var size = dependentResourcesByName.size();
+    if (size > 0) {
+      dependentResourcesByName.forEach((key, value) -> {
+        if (value instanceof EventSourceProvider) {
+          final var provider = (EventSourceProvider) value;
+          final var source = provider.initEventSource(context);
+          eventSourceManager.registerEventSource(key, source);
+        } else {
+          Optional<ResourceEventSource> eventSource = value.eventSource(context);
+          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(key, es));
+        }
+      });
+
+      // resolve event sources referenced by name for dependents that reuse an existing event source
+      final Map<String, List<DeferrableEventSourceHolder>> unresolvable = new HashMap<>(size);
+      dependentResourcesByName.values().stream()
+          .filter(DeferrableEventSourceHolder.class::isInstance)
+          .map(DeferrableEventSourceHolder.class::cast)
+          .forEach(dr -> ((DeferrableEventSourceHolder<P>) dr)
+              .resolveEventSource(eventSourceManager).ifPresent(unresolved -> unresolvable
+                  .computeIfAbsent(unresolved, s -> new ArrayList<>()).add(dr)));
+      if (!unresolvable.isEmpty()) {
+        throw new IllegalStateException(
+            "Couldn't resolve referenced EventSources: " + unresolvable);
+      }
+    }
   }
 
   @Override

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

@@ -5,7 +5,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceAware;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DeferrableEventSourceHolder;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
 import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -17,7 +17,7 @@
 
 @Ignore
 public abstract class AbstractEventSourceHolderDependentResource<R, P extends HasMetadata, T extends ResourceEventSource<R, P>>
-    extends AbstractDependentResource<R, P> implements EventSourceAware<P> {
+    extends AbstractDependentResource<R, P> implements DeferrableEventSourceHolder<P> {
 
   private T eventSource;
   private final Class<R> resourceType;
@@ -32,7 +32,6 @@ protected AbstractEventSourceHolderDependentResource(Class<R> resourceType) {
     this.resourceType = resourceType;
   }
 
-
   public Optional<ResourceEventSource<R, P>> eventSource(EventSourceContext<P> context) {
     // some sub-classes (e.g. KubernetesDependentResource) can have their event source created
     // before this method is called in the managed case, so only create the event source if it
@@ -49,14 +48,20 @@ public Optional<ResourceEventSource<R, P>> eventSource(EventSourceContext<P> con
 
   @SuppressWarnings("unchecked")
   @Override
-  public void selectEventSources(EventSourceRetriever<P> eventSourceRetriever) {
-    if (eventSourceNameToUse != null) {
-      setEventSource(
-          (T) eventSourceRetriever.getResourceEventSourceFor(resourceType(), eventSourceNameToUse));
+  public Optional<String> resolveEventSource(EventSourceRetriever<P> eventSourceRetriever) {
+    if (eventSourceNameToUse != null && eventSource == null) {
+      final var source =
+          eventSourceRetriever.getResourceEventSourceFor(resourceType(), eventSourceNameToUse);
+      if (source == null) {
+        return Optional.of(eventSourceNameToUse);
+      }
+      setEventSource((T) source);
     }
+    return Optional.empty();
   }
 
   /** To make this backwards compatible even for respect of overriding */
+  @SuppressWarnings("unchecked")
   public T initEventSource(EventSourceContext<P> context) {
     return (T) eventSource(context).orElseThrow();
   }
@@ -117,10 +122,4 @@ public void setOnUpdateFilter(OnUpdateFilter<R> onUpdateFilter) {
   public void setOnDeleteFilter(OnDeleteFilter<R> onDeleteFilter) {
     this.onDeleteFilter = onDeleteFilter;
   }
-
-  public AbstractEventSourceHolderDependentResource<R, P, T> setEventSourceNameToUse(
-      String eventSourceNameToUse) {
-    this.eventSourceNameToUse = eventSourceNameToUse;
-    return this;
-  }
 }

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

@@ -14,6 +14,7 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DeferrableEventSourceHolder;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DependentResourceConfigurator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.KubernetesClientAware;
@@ -83,7 +84,17 @@ public DependentResource createAndConfigureFrom(DependentResourceSpec spec,
     }
 
     spec.getUseEventSourceWithName()
-        .ifPresent(esName -> dependentResource.useEventSourceWithName((String) esName));
+        .ifPresent(esName -> {
+          final var name = (String) esName;
+          if (dependentResource instanceof DeferrableEventSourceHolder) {
+            ((DeferrableEventSourceHolder) dependentResource).useEventSourceWithName(name);
+          } else {
+            throw new IllegalStateException(
+                "DependentResource " + spec + " wants to use EventSource named " + name
+                    + " but doesn't implement support for this feature by implementing "
+                    + DeferrableEventSourceHolder.class.getSimpleName());
+          }
+        });
 
     if (dependentResource instanceof DependentResourceConfigurator) {
       final var configurator = (DependentResourceConfigurator) dependentResource;