fix: PerResourcePollingEventSource issue with resources

Author: csviri

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

@@ -134,8 +134,8 @@ public final synchronized void registerEventSource(String name, EventSource even
   @SuppressWarnings("unchecked")
   public void broadcastOnResourceEvent(ResourceAction action, R resource, R oldResource) {
     eventSources.additionalNamedEventSources().forEach(eventSource -> {
-      if (eventSource instanceof ResourceEventAware) {
-        var lifecycleAwareES = ((ResourceEventAware<R>) eventSource);
+      if (eventSource.original() instanceof ResourceEventAware) {
+        var lifecycleAwareES = ((ResourceEventAware<R>) eventSource.original());
         switch (action) {
           case ADDED:
             lifecycleAwareES.onResourceCreated(resource);

operator-framework/src/test/java/io/javaoperatorsdk/operator/PerResourcePollingEventSourceIT.java

@@ -0,0 +1,42 @@
+package io.javaoperatorsdk.operator;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.sample.perresourceeventsource.PerResourceEventSourceCustomResource;
+import io.javaoperatorsdk.operator.sample.perresourceeventsource.PerResourcePollingEventSourceTestReconciler;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class PerResourcePollingEventSourceIT {
+
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(new PerResourcePollingEventSourceTestReconciler())
+          .build();
+
+  @Test
+  void managedDependentsAreReconciledInOrder() {
+    operator.create(PerResourceEventSourceCustomResource.class, resource());
+
+    var reconciler =
+        operator.getReconcilerOfType(PerResourcePollingEventSourceTestReconciler.class);
+    await().untilAsserted(() -> {
+      assertThat(reconciler.getNumberOfExecutions()).isGreaterThan(2);
+      assertThat(reconciler.getNumberOfFetchExecution()).isGreaterThan(2);
+    });
+  }
+
+  private PerResourceEventSourceCustomResource resource() {
+    var res = new PerResourceEventSourceCustomResource();
+    res.setMetadata(new ObjectMetaBuilder()
+        .withName("test1")
+        .build());
+    return res;
+  }
+
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourceEventSourceCustomResource.java

@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.perresourceeventsource;
+
+import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.client.CustomResource;
+import io.fabric8.kubernetes.model.annotation.Group;
+import io.fabric8.kubernetes.model.annotation.ShortNames;
+import io.fabric8.kubernetes.model.annotation.Version;
+
+@Group("sample.javaoperatorsdk")
+@Version("v1")
+@ShortNames("pres")
+public class PerResourceEventSourceCustomResource
+    extends CustomResource<Void, Void>
+    implements Namespaced {
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java

@@ -0,0 +1,65 @@
+package io.javaoperatorsdk.operator.sample.perresourceeventsource;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import io.javaoperatorsdk.operator.junit.KubernetesClientAware;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource;
+
+@ControllerConfiguration
+public class PerResourcePollingEventSourceTestReconciler
+    implements Reconciler<PerResourceEventSourceCustomResource>,
+    EventSourceInitializer<PerResourceEventSourceCustomResource>,
+    KubernetesClientAware {
+
+  public static final int POLL_PERIOD = 100;
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+  private final AtomicInteger numberOfFetchExecutions = new AtomicInteger(0);
+
+  private KubernetesClient client;
+  private PerResourcePollingEventSource<String, PerResourceEventSourceCustomResource> eventSource;
+
+  @Override
+  public UpdateControl<PerResourceEventSourceCustomResource> reconcile(
+      PerResourceEventSourceCustomResource resource,
+      Context<PerResourceEventSourceCustomResource> context) throws Exception {
+    numberOfExecutions.addAndGet(1);
+    return UpdateControl.noUpdate();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(
+      EventSourceContext<PerResourceEventSourceCustomResource> context) {
+    this.eventSource =
+        new PerResourcePollingEventSource<>(primaryResource -> {
+          numberOfFetchExecutions.addAndGet(1);
+          return Set.of(UUID.randomUUID().toString());
+        },
+            context.getPrimaryCache(), POLL_PERIOD, String.class);
+    return EventSourceInitializer.nameEventSources(eventSource);
+  }
+
+  @Override
+  public KubernetesClient getKubernetesClient() {
+    return client;
+  }
+
+  @Override
+  public void setKubernetesClient(KubernetesClient kubernetesClient) {
+    this.client = kubernetesClient;
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  public int getNumberOfFetchExecution() {
+    return numberOfFetchExecutions.get();
+  }
+
+}