diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java index af9c3c8d9f..2e8c6d0aad 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java +++ b/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

context) { managedWorkflow.getDependentResourcesWithoutActivationCondition(); final var size = dependentResourcesByName.size(); if (size > 0) { - dependentResourcesByName.forEach(dependentResource -> { - Optional eventSource = dependentResource.eventSource(context); - eventSource.ifPresent(eventSourceManager::registerEventSource); + dependentResourcesByName.forEach(dr -> { + Optional 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, Resource

> getCRClient() { + public MixedOperation, Resource

> getPrimaryResourceClient() { return kubernetesClient.resources(configuration.getResourceClass()); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java index 2bb8c7a39a..8bf3e65f4b 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java @@ -213,9 +213,9 @@ public List> getEventSourcesFor(Class dependentType) { @Override public EventSource dynamicallyRegisterEventSource(EventSource eventSource) { synchronized (this) { - var actual = eventSources.existingEventSourceByName(eventSource.name()); + var actual = eventSources.getEventSourceByName(eventSource.name()); if (actual != null) { - eventSource = actual; + eventSource = (EventSource) actual; } else { registerEventSource(eventSource); } @@ -249,6 +249,11 @@ public EventSource getEventSourceFor( return eventSources.get(dependentType, name); } + + public Optional> getEventSourceByName(String name) { + return Optional.ofNullable(eventSources.getEventSourceByName(name)); + } + TimerEventSource

retryEventSource() { return eventSources.retryEventSource(); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java index e790ae3c32..f1d79f7ca2 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java @@ -19,13 +19,13 @@ class EventSources

{ private final ConcurrentNavigableMap>> sources = new ConcurrentSkipListMap<>(); - private final Map sourceByName = new HashMap<>(); + private final Map> sourceByName = new HashMap<>(); private final TimerEventSource

retryAndRescheduleTimerEventSource = new TimerEventSource<>("RetryAndRescheduleTimerEventSource"); private ControllerEventSource

controllerEventSource; - public void add(EventSource eventSource) { + public void add(EventSource 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 getEventSourceByName(String name) { return sourceByName.get(name); } @@ -92,15 +92,15 @@ private String keyFor(Class dependentType) { } @SuppressWarnings("unchecked") - public EventSource get(Class dependentType, String name) { - if (dependentType == null) { + public EventSource get(Class 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,7 +110,7 @@ public EventSource get(Class 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())); } @@ -118,16 +118,16 @@ public EventSource get(Class dependentType, String 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; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java index 1ad3b65910..97086c14f7 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java @@ -58,7 +58,8 @@ class ReconciliationDispatcher

{ public ReconciliationDispatcher(Controller

controller) { this(controller, - new CustomResourceFacade<>(controller.getCRClient(), controller.getConfiguration())); + new CustomResourceFacade<>(controller.getPrimaryResourceClient(), + controller.getConfiguration())); } public PostExecutionControl

handleExecution(ExecutionScope

executionScope) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java index db457a4e41..460d4dfa45 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java @@ -32,7 +32,7 @@ public class ControllerEventSource @SuppressWarnings({"unchecked", "rawtypes"}) public ControllerEventSource(Controller controller) { - super(NAME, controller.getCRClient(), controller.getConfiguration(), false); + super(NAME, controller.getPrimaryResourceClient(), controller.getConfiguration(), false); this.controller = controller; final var config = controller.getConfiguration(); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java index e97cf78b1d..8bc62c0a57 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java @@ -36,12 +36,6 @@ protected ConfigMap desired(CreateUpdateEventFilterTestCustomResource primary, Context context) { return desired; } - - @Override - public void setEventSource( - io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource eventSource) { - super.setEventSource(eventSource); - } } public static final String CONFIG_MAP_TEST_DATA_KEY = "key";