fix: stop timers when closing TimerEventSources

Author: lburgazzoli

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

@@ -3,23 +3,28 @@
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.processing.KubernetesResourceUtils;
 import io.javaoperatorsdk.operator.processing.event.AbstractEventSource;
+import java.io.IOException;
 import java.util.Map;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class TimerEventSource extends AbstractEventSource {
 
-  private Logger log = LoggerFactory.getLogger(TimerEventSource.class);
-
   private final Timer timer = new Timer();
-
+  private final AtomicBoolean running = new AtomicBoolean();
   private final Map<String, EventProducerTimeTask> onceTasks = new ConcurrentHashMap<>();
   private final Map<String, EventProducerTimeTask> timerTasks = new ConcurrentHashMap<>();
+  private Logger log = LoggerFactory.getLogger(TimerEventSource.class);
 
   public void schedule(CustomResource customResource, long delay, long period) {
+    if (!running.get()) {
+      throw new IllegalStateException("The TimerEventSource is not running");
+    }
+
     String resourceUid = KubernetesResourceUtils.getUID(customResource);
     if (timerTasks.containsKey(resourceUid)) {
       return;
@@ -30,6 +35,10 @@ public void schedule(CustomResource customResource, long delay, long period) {
   }
 
   public void scheduleOnce(CustomResource customResource, long delay) {
+    if (!running.get()) {
+      throw new IllegalStateException("The TimerEventSource is not running");
+    }
+
     String resourceUid = KubernetesResourceUtils.getUID(customResource);
     if (onceTasks.containsKey(resourceUid)) {
       cancelOnceSchedule(resourceUid);
@@ -59,6 +68,19 @@ public void cancelOnceSchedule(String customResourceUid) {
     }
   }
 
+  @Override
+  public void start() {
+    running.set(true);
+  }
+
+  @Override
+  public void close() throws IOException {
+    running.set(false);
+    onceTasks.keySet().forEach(this::cancelOnceSchedule);
+    timerTasks.keySet().forEach(this::cancelSchedule);
+    timer.cancel();
+  }
+
   public class EventProducerTimeTask extends TimerTask {
 
     protected final String customResourceUid;
@@ -69,8 +91,10 @@ public EventProducerTimeTask(String customResourceUid) {
 
     @Override
     public void run() {
-      log.debug("Producing event for custom resource id: {}", customResourceUid);
-      eventHandler.handleEvent(new TimerEvent(customResourceUid, TimerEventSource.this));
+      if (running.get()) {
+        log.debug("Producing event for custom resource id: {}", customResourceUid);
+        eventHandler.handleEvent(new TimerEvent(customResourceUid, TimerEventSource.this));
+      }
     }
   }
 }

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

@@ -2,6 +2,7 @@
 
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -13,6 +14,7 @@
 import io.javaoperatorsdk.operator.TestUtils;
 import io.javaoperatorsdk.operator.processing.KubernetesResourceUtils;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
+import java.io.IOException;
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
@@ -33,6 +35,7 @@ class TimerEventSourceTest {
   public void setup() {
     timerEventSource = new TimerEventSource();
     timerEventSource.setEventHandler(eventHandlerMock);
+    timerEventSource.start();
   }
 
   @Test
@@ -106,4 +109,23 @@ public void deRegistersOnceEventSources() throws InterruptedException {
 
     verify(eventHandlerMock, never()).handleEvent(any());
   }
+
+  @Test
+  public void eventNotRegisteredIfStopped() throws IOException {
+    CustomResource customResource = TestUtils.testCustomResource();
+
+    timerEventSource.close();
+    assertThatExceptionOfType(IllegalStateException.class).isThrownBy(
+        () -> timerEventSource.scheduleOnce(customResource, PERIOD));
+  }
+
+  @Test
+  public void eventNotFiredIfStopped() throws InterruptedException, IOException {
+    timerEventSource.scheduleOnce(TestUtils.testCustomResource(), PERIOD);
+    timerEventSource.close();
+
+    Thread.sleep(PERIOD + TESTING_TIME_SLACK);
+
+    verify(eventHandlerMock, never()).handleEvent(any());
+  }
 }