wip

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

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

@@ -140,21 +140,26 @@ private boolean isAcceptedByFilters(ResourceAction action, T resource, T oldReso
 
   @Override
   public synchronized void onAdd(T resource) {
-    var handling = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, resource, null);
-    handleEvent(ResourceAction.ADDED, resource, null, null, handling != EventHandling.NEW);
+    handleOnAddOrUpdate(ResourceAction.ADDED, null, resource);
   }
 
   @Override
   public synchronized void onUpdate(T oldCustomResource, T newCustomResource) {
+    handleOnAddOrUpdate(ResourceAction.UPDATED, oldCustomResource, newCustomResource);
+  }
+
+  private void handleOnAddOrUpdate(
+      ResourceAction action, T oldCustomResource, T newCustomResource) {
     var handling =
-        temporaryResourceCache.onAddOrUpdateEvent(
-            ResourceAction.UPDATED, newCustomResource, oldCustomResource);
-    handleEvent(
-        ResourceAction.UPDATED,
-        newCustomResource,
-        oldCustomResource,
-        null,
-        handling != EventHandling.NEW);
+        temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource);
+    if (handling != EventHandling.NEW) {
+      handleEvent(
+          ResourceAction.UPDATED,
+          newCustomResource,
+          oldCustomResource,
+          null,
+          handling != EventHandling.NEW);
+    }
   }
 
   @Override

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java

@@ -23,6 +23,7 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.javaoperatorsdk.operator.MockKubernetesClient;
 import io.javaoperatorsdk.operator.ReconcilerUtilsInternal;
 import io.javaoperatorsdk.operator.TestUtils;
@@ -44,6 +45,8 @@
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static io.javaoperatorsdk.operator.processing.event.source.EventFilterTestUtils.withResourceVersion;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.*;
 
@@ -168,17 +171,86 @@ void genericFilterFiltersOutAddUpdateAndDeleteEvents() {
     verify(eventHandler, never()).handleEvent(any());
   }
 
-  @Disabled
   @Test
-  void testEventFiltering() throws InterruptedException {
+  void testEventFilteringBasicScenario() throws InterruptedException {
     source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
     setUpSource(source, true, controllerConfig);
 
     var latch = sendForEventFilteringUpdate(2);
     source.onUpdate(testResourceWithVersion(1), testResourceWithVersion(2));
     latch.countDown();
+
     Thread.sleep(100);
-    verify(source, never()).handleEvent(any(), any(), any(), any(), anyBoolean());
+    verify(eventHandler, never()).handleEvent(any());
+  }
+
+  @Test
+  void eventFilteringNewEventDuringUpdate() {
+    source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
+    setUpSource(source, true, controllerConfig);
+
+    var latch = sendForEventFilteringUpdate(2);
+    source.onUpdate(testResourceWithVersion(2), testResourceWithVersion(3));
+    latch.countDown();
+
+    await().untilAsserted(() -> expectHandleEvent(3, 2));
+  }
+
+  @Disabled("todo")
+  @Test
+  void eventFilteringMoreNewEventsDuringUpdate() {
+    source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
+    setUpSource(source, true, controllerConfig);
+
+    var latch = sendForEventFilteringUpdate(2);
+    source.onUpdate(testResourceWithVersion(2), testResourceWithVersion(3));
+    source.onUpdate(testResourceWithVersion(3), testResourceWithVersion(4));
+    latch.countDown();
+
+    await().untilAsserted(() -> expectHandleEvent(4, 2));
+  }
+
+  @Test
+  void eventFilteringExceptionDuringUpdate() {
+    source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
+    setUpSource(source, true, controllerConfig);
+
+    var latch =
+        EventFilterTestUtils.sendForEventFilteringUpdate(
+            source,
+            TestUtils.testCustomResource1(),
+            r -> {
+              throw new KubernetesClientException("fake");
+            });
+    source.onUpdate(testResourceWithVersion(1), testResourceWithVersion(2));
+    latch.countDown();
+
+    expectHandleEvent(2, 1);
+  }
+
+  private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) {
+    await()
+        .untilAsserted(
+            () -> {
+              verify(eventHandler, times(1)).handleEvent(any());
+              verify(source, times(1))
+                  .handleEvent(
+                      eq(ResourceAction.UPDATED),
+                      argThat(
+                          r -> {
+                            assertThat(r.getMetadata().getResourceVersion())
+                                .isEqualTo("" + newResourceVersion);
+                            return true;
+                          }),
+                      argThat(
+                          r -> {
+                            assertThat(r.getMetadata().getResourceVersion())
+                                .isEqualTo("" + oldResourceVersion);
+                            return true;
+                          }),
+                      isNull(),
+                      eq(false));
+            });
   }
 
   private TestCustomResource testResourceWithVersion(int v) {