#3024 removing any onClose action

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

CHANGELOG.md

@@ -14,6 +14,7 @@
 * Fix #3011: properly handle enum types for additional printer columns
 * Fix #3020: annotations should now properly have their associated values when processing CRDs from the API
 * Fix #3027: fix NPE when sorting events in KubernetesResourceUtil
+* 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

@@ -63,7 +63,7 @@ public class Controller<T extends HasMetadata, L extends KubernetesResourceList<
 
   private final ScheduledExecutorService resyncExecutor;
 
-  private ScheduledFuture resyncFuture;
+  private volatile ScheduledFuture resyncFuture;
 
   private final OperationContext operationContext;
 
@@ -105,7 +105,8 @@ public void run() {
 
     try {
       running = true;
-      reflector.listAndWatch();
+      log.info("Started Reflector watch for {}", apiTypeClass);
+      reflector.listSyncAndWatch();
 
       // Start the process loop
       this.processLoop();
@@ -121,13 +122,11 @@ public void run() {
    * Stops the resync thread pool first, then stops the reflector.
    */
   public void stop() {
-    synchronized (this) {
-      reflector.stop();
-      if (resyncFuture != null) {
-        resyncFuture.cancel(true);
-      }
-      resyncExecutor.shutdown();
+    reflector.stop();
+    if (resyncFuture != null) {
+      resyncFuture.cancel(true);
     }
+    resyncExecutor.shutdown();
   }
 
   /**

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

@@ -30,7 +30,6 @@
 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;
@@ -47,7 +46,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.watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::listSyncAndWatch);
     this.isActive = new AtomicBoolean(true);
     this.watch = new AtomicReference<>(null);
   }
@@ -59,54 +58,48 @@ private L getList() {
       .withTimeoutSeconds(null).build(), operationContext.getNamespace(), operationContext);
   }
 
-  public void listAndWatch() {
-    log.info("Started ReflectorRunnable watch for {}", apiTypeClass);
-    reListAndSync();
-    startWatcher();
-  }
-
   public void stop() {
     isActive.set(false);
+    stopWatcher();
+  }
+
+  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 (!isActive.get()) {
         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();
+      return lastSyncResourceVersion.get();
   }
 
   public boolean isRunning() {

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