Fix #3024 removing any onClose action (#3055)

* it's not necessary given the retrying that happens from the watch
framework

* switching to not even track the future to pass sonar check

* switching to a stopped boolean

* changing to running

* removing changes that could lead to conflicts

CHANGELOG.md

@@ -19,6 +19,7 @@
 * Fix #3027: fix NPE when sorting events in KubernetesResourceUtil
 * Fix missing entry for Trigger in TektonTriggersResourceMappingProvider
 * Fix #3047: NPE when getting version when there is no build date
+* Fix #3024: stopAllRegisteredInformers will not call startWatcher
 
 #### Improvements
 * Fix #2788: Support FIPS mode in kubernetes-client with BouncyCastleFipsProvider

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/cache/Controller.java

@@ -102,7 +102,8 @@ public void run() {
 
     try {
       running = true;
-      reflector.listAndWatch();
+      log.info("Started Reflector watch for {}", apiTypeClass);
+      reflector.listSyncAndWatch();
 
       // Start the process loop
       this.processLoop();

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/cache/Reflector.java

@@ -24,21 +24,19 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class Reflector<T extends HasMetadata, L extends KubernetesResourceList<T>> {
 
   private static final Logger log = LoggerFactory.getLogger(Reflector.class);
-  private static final Long WATCH_RESTART_DELAY_MILLIS = 5000L;
 
   private final AtomicReference<String> lastSyncResourceVersion;
   private final Class<T> apiTypeClass;
   private final ListerWatcher<T, L> listerWatcher;
   private final Store<T> store;
   private final OperationContext operationContext;
   private final ReflectorWatcher<T> watcher;
-  private final AtomicBoolean isActive;
+  private volatile boolean running = true;
   private final AtomicReference<Watch> watch;
 
   public Reflector(Class<T> apiTypeClass, ListerWatcher<T, L> listerWatcher, Store store, OperationContext operationContext) {
@@ -47,8 +45,7 @@ public Reflector(Class<T> apiTypeClass, ListerWatcher<T, L> listerWatcher, Store
     this.store = store;
     this.operationContext = operationContext;
     this.lastSyncResourceVersion = new AtomicReference<>();
-    this.watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::startWatcher, this::reListAndSync);
-    this.isActive = new AtomicBoolean(true);
+    this.watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::listSyncAndWatch);
     this.watch = new AtomicReference<>(null);
   }
 
@@ -59,57 +56,51 @@ private L getList() {
       .withTimeoutSeconds(null).build(), operationContext.getNamespace(), operationContext);
   }
 
-  public void listAndWatch() {
-    log.info("Started ReflectorRunnable watch for {}", apiTypeClass);
-    reListAndSync();
-    startWatcher();
+  public void stop() {
+    running = false;
+    stopWatcher();
   }
 
-  public void stop() {
-    isActive.set(false);
+  private synchronized void stopWatcher() {
     Watch theWatch = watch.getAndSet(null);
     if (theWatch != null) {
+      String ns = operationContext.getNamespace();
+      log.debug("Stopping watcher for resource {} v{} in namespace {}", apiTypeClass, lastSyncResourceVersion.get(), ns);
       theWatch.close();
     }
   }
 
-  private void reListAndSync() {
+  /**
+   * <br>Starts the watch with a fresh store state.
+   * <br>Should be called only at start and when HttpGone is seen.
+   */
+  void listSyncAndWatch() {
     store.isPopulated(false);
     final L list = getList();
     final String latestResourceVersion = list.getMetadata().getResourceVersion();
+    lastSyncResourceVersion.set(list.getMetadata().getResourceVersion());
     log.debug("Listing items ({}) for resource {} v{}", list.getItems().size(), apiTypeClass, latestResourceVersion);
-    lastSyncResourceVersion.set(latestResourceVersion);
     store.replace(list.getItems(), latestResourceVersion);
+    startWatcher(latestResourceVersion);
   }
 
-  private void startWatcher() {
-    log.debug("Starting watcher for resource {} v{}", apiTypeClass, lastSyncResourceVersion.get());
-    Watch theWatch = watch.getAndSet(null);
-    if (theWatch != null) {
-      theWatch.close();
-      log.debug("Stopping previous watcher, and delaying execution of new watcher");
-      try {
-        Thread.sleep(WATCH_RESTART_DELAY_MILLIS);
-      } catch (InterruptedException e) {
-        log.debug("Reflector thread was interrupted: {}", e.getMessage());
-        Thread.currentThread().interrupt();
+  private synchronized void startWatcher(final String latestResourceVersion) {
+    if (!running) {
         return;
-      }
-    }
-    if (isActive.get()) {
-      watch.set(
-        listerWatcher.watch(new ListOptionsBuilder()
-          .withWatch(Boolean.TRUE).withResourceVersion(lastSyncResourceVersion.get()).withTimeoutSeconds(null).build(),
-        operationContext.getNamespace(), operationContext, watcher)
-      );
     }
+    log.debug("Starting watcher for resource {} v{}", apiTypeClass, latestResourceVersion);
+    // there's no need to stop the old watch, that will happen automatically when this call completes
+    watch.set(
+      listerWatcher.watch(new ListOptionsBuilder()
+        .withWatch(Boolean.TRUE).withResourceVersion(latestResourceVersion).withTimeoutSeconds(null).build(),
+      operationContext.getNamespace(), operationContext, watcher));
   }
 
   public String getLastSyncResourceVersion() {
     return lastSyncResourceVersion.get();
   }
 
   public boolean isRunning() {
-    return isActive.get();
+    return running;
   }
 }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/cache/ReflectorWatcher.java

@@ -30,13 +30,11 @@ public class ReflectorWatcher<T extends HasMetadata> implements Watcher<T> {
 
   private final Store<T> store;
   private final AtomicReference<String> lastSyncResourceVersion;
-  private final Runnable onClose;
   private final Runnable onHttpGone;
 
-  public ReflectorWatcher(Store<T> store, AtomicReference<String> lastSyncResourceVersion, Runnable onClose, Runnable onHttpGone) {
+  public ReflectorWatcher(Store<T> store, AtomicReference<String> lastSyncResourceVersion, Runnable onHttpGone) {
     this.store = store;
     this.lastSyncResourceVersion = lastSyncResourceVersion;
-    this.onClose = onClose;
     this.onHttpGone = onHttpGone;
   }
 
@@ -75,13 +73,11 @@ public void onClose(WatcherException exception) {
     if (exception.isHttpGone()) {
         onHttpGone.run();
     }
-    onClose.run();
   }
 
   @Override
   public void onClose() {
     log.debug("Watch gracefully closed");
-    onClose.run();
   }
 
   @Override