fix: DefaultEventHandler should not fire events if the handler is being closed #578

Author: lburgazzoli

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

@@ -51,6 +51,7 @@ public void failedEvent(String uid, Event event) {}
   private final ExecutorService executor;
   private final String controllerName;
   private final ReentrantLock lock = new ReentrantLock();
+  private volatile boolean running;
   private DefaultEventSourceManager<R> eventSourceManager;
 
   public DefaultEventHandler(ConfiguredController<R> controller) {
@@ -67,6 +68,7 @@ public DefaultEventHandler(ConfiguredController<R> controller) {
 
   private DefaultEventHandler(ExecutorService executor, String relatedControllerName,
       EventDispatcher<R> eventDispatcher, Retry retry) {
+    this.running = true;
     this.executor =
         executor == null
             ? new ScheduledThreadPoolExecutor(
@@ -75,27 +77,27 @@ private DefaultEventHandler(ExecutorService executor, String relatedControllerNa
     this.controllerName = relatedControllerName;
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
-    eventBuffer = new EventBuffer();
-  }
-
-  public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
-    this.eventSourceManager = eventSourceManager;
+    this.eventBuffer = new EventBuffer();
   }
 
   public static void setEventMonitor(EventMonitor monitor) {
     DefaultEventHandler.monitor = monitor;
   }
 
-  public interface EventMonitor {
-    void processedEvent(String uid, Event event);
-
-    void failedEvent(String uid, Event event);
+  public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
+    this.eventSourceManager = eventSourceManager;
   }
 
   @Override
   public void handleEvent(Event event) {
+
     try {
       lock.lock();
+
+      if (!this.running) {
+        log.debug("Skipping event: {} because the event handler is shutting down", event);
+      }
+
       log.debug("Received event: {}", event);
 
       final Predicate<CustomResource> selector = event.getCustomResourcesSelector();
@@ -109,6 +111,16 @@ public void handleEvent(Event event) {
     }
   }
 
+  @Override
+  public void close() {
+    try {
+      lock.lock();
+      this.running = false;
+    } finally {
+      lock.unlock();
+    }
+  }
+
   private void executeBufferedEvents(String customResourceUid) {
     boolean newEventForResourceId = eventBuffer.containsEvents(customResourceUid);
     boolean controllerUnderExecution = isControllerUnderExecution(customResourceUid);
@@ -143,6 +155,10 @@ void eventProcessingFinished(
       ExecutionScope<R> executionScope, PostExecutionControl<R> postExecutionControl) {
     try {
       lock.lock();
+      if (!running) {
+        return;
+      }
+
       log.debug(
           "Event processing finished. Scope: {}, PostExecutionControl: {}",
           executionScope,
@@ -279,6 +295,12 @@ private void unsetUnderExecution(String customResourceUid) {
     underProcessing.remove(customResourceUid);
   }
 
+  public interface EventMonitor {
+    void processedEvent(String uid, Event event);
+
+    void failedEvent(String uid, Event event);
+  }
+
   private class ControllerExecution implements Runnable {
     private final ExecutionScope<R> executionScope;
 

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

@@ -54,6 +54,13 @@ public DefaultEventSourceManager(ConfiguredController<R> controller) {
   public void close() {
     try {
       lock.lock();
+
+      try {
+        defaultEventHandler.close();
+      } catch (Exception e) {
+        log.warn("Error closing event handler", e);
+      }
+
       for (var entry : eventSources.entrySet()) {
         try {
           log.debug("Closing {} -> {}", entry.getKey(), entry.getValue());

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

@@ -23,7 +23,15 @@
 import static io.javaoperatorsdk.operator.TestUtils.testCustomResource;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 class DefaultEventHandlerTest {
 
@@ -224,6 +232,14 @@ public void scheduleTimedEventIfInstructedByPostExecutionControl() {
         .scheduleOnce(any(), eq(testDelay));
   }
 
+  @Test
+  public void doNotFireEventsIfClosing() {
+    defaultEventHandler.close();
+    defaultEventHandler.handleEvent(prepareCREvent());
+
+    verify(eventDispatcherMock, timeout(50).times(1)).handleExecution(any());
+  }
+
   private void waitMinimalTime() {
     try {
       Thread.sleep(50);