fix: restart event handler (#632)

Properly start event handler when starting the event source. 
Minor clean-ups.
Fixes #630
Co-authored-by: csviri <[email protected]>

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java

@@ -46,7 +46,7 @@ public static void close() {
   public static ExecutorServiceManager instance() {
     if (instance == null) {
       throw new IllegalStateException(
-          "ExecutorServiceManager hasn't been started. Call start method before using!");
+          "ExecutorServiceManager hasn't been started. Call init method before using!");
     }
     return instance;
   }

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

@@ -60,11 +60,15 @@ public DefaultEventHandler(ConfiguredController<R> controller) {
         controller.getConfiguration().getConfigurationService().getMetrics().getEventMonitor());
   }
 
-  DefaultEventHandler(EventDispatcher<R> eventDispatcher, String relatedControllerName,
+  public DefaultEventHandler(EventDispatcher<R> eventDispatcher, String relatedControllerName,
       Retry retry) {
     this(null, relatedControllerName, eventDispatcher, retry, null);
   }
 
+  public boolean isRunning() {
+    return running;
+  }
+
   private DefaultEventHandler(ExecutorService executor, String relatedControllerName,
       EventDispatcher<R> eventDispatcher, Retry retry, EventMonitor monitor) {
     this.running = true;
@@ -142,6 +146,15 @@ public void handleEvent(Event event) {
     }
   }
 
+  public void start() {
+    try {
+      lock.lock();
+      this.running = true;
+    } finally {
+      lock.unlock();
+    }
+  }
+
   @Override
   public void close() {
     try {

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

@@ -9,4 +9,6 @@ public interface EventHandler extends Closeable {
 
   @Override
   default void close() throws IOException {}
+
+  default void start() {}
 }

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

@@ -48,6 +48,8 @@ public CustomResourceEventSource(ConfiguredController<T> controller) {
 
   @Override
   public void start() {
+    eventHandler.start();
+
     final var configuration = controller.getConfiguration();
     final var targetNamespaces = configuration.getEffectiveNamespaces();
     final var client = controller.getCRClient();

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java

@@ -63,7 +63,6 @@ public void setup() {
     // todo: remove
     when(defaultEventSourceManagerMock.getCache()).thenReturn(customResourceCache);
     doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResource(any());
-    doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResource(any());
     doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResources(any());
     doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResourceUids(any());
     doCallRealMethod().when(defaultEventSourceManagerMock).cacheResource(any(), any());

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventSourceTest.java

@@ -1,25 +1,37 @@
 package io.javaoperatorsdk.operator.processing.event.internal;
 
+import java.io.IOException;
 import java.time.LocalDateTime;
 import java.util.List;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import io.fabric8.kubernetes.api.model.KubernetesResourceList;
+import io.fabric8.kubernetes.api.model.ListOptions;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.dsl.FilterWatchListMultiDeletable;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.Metrics;
 import io.javaoperatorsdk.operator.TestUtils;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.DefaultControllerConfiguration;
 import io.javaoperatorsdk.operator.processing.ConfiguredController;
+import io.javaoperatorsdk.operator.processing.CustomResourceCache;
+import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
+import io.javaoperatorsdk.operator.processing.EventDispatcher;
+import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -102,6 +114,46 @@ public void eventNotMarkedForLastGenerationIfNoFinalizer() {
     verify(eventHandler, times(2)).handleEvent(any());
   }
 
+  @Test
+  public void restartingShouldResumeEventHandling() throws IOException {
+    final var cr = TestUtils.testCustomResource();
+
+    CustomResourceCache customResourceCache = new CustomResourceCache();
+    customResourceCache.cacheResource(cr);
+    DefaultEventSourceManager defaultEventSourceManagerMock =
+        mock(DefaultEventSourceManager.class);
+    EventDispatcher eventDispatcherMock = mock(EventDispatcher.class);
+    DefaultEventHandler local = new DefaultEventHandler(eventDispatcherMock, "Test",
+        null);
+    local.setEventSourceManager(defaultEventSourceManagerMock);
+    when(defaultEventSourceManagerMock.getCache()).thenReturn(customResourceCache);
+    doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResource(any());
+    doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResources(any());
+    doCallRealMethod().when(defaultEventSourceManagerMock).getLatestResourceUids(any());
+    doCallRealMethod().when(defaultEventSourceManagerMock).cacheResource(any(), any());
+
+    final var mock = mock(FilterWatchListMultiDeletable.class);
+    when(mock.watch((ListOptions) any(), any())).thenReturn(mock(Watch.class));
+    when(client.inAnyNamespace()).thenReturn(mock);
+
+    customResourceEventSource.setEventHandler(local);
+
+    customResourceEventSource.eventReceived(Watcher.Action.MODIFIED, cr);
+    verify(eventDispatcherMock, timeout(50).times(1)).handleExecution(any());
+
+    customResourceEventSource.close();
+    assertFalse(local.isRunning());
+    customResourceEventSource.eventReceived(Watcher.Action.MODIFIED, cr);
+    // mockito times method is not reset and keeps increasing so here we stay at 1 call
+    verify(eventDispatcherMock, timeout(50).times(1)).handleExecution(any());
+
+    customResourceEventSource.start();
+    assertTrue(local.isRunning());
+    customResourceEventSource.eventReceived(Watcher.Action.MODIFIED, cr);
+    // we're expecting another call to the dispatcher, so total number of calls should now be 2
+    verify(eventDispatcherMock, timeout(50).times(2)).handleExecution(any());
+  }
+
   private static class TestConfiguredController extends ConfiguredController<TestCustomResource> {
 
     public TestConfiguredController(boolean generationAware) {