Change the onClose method of ReflectorWatcher instance to only close …

…watcher and not re-open it

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 5.4-SNAPSHOT
 
 #### Bugs
+* Fix #3024: stopAllRegisteredInformers will not call startWatcher but stopWatcher onClose instead
 * Fix #2989: serialization will generate valid yaml when using subtypes
 * Fix #2996: Generating CRDs from the API should now properly work
 * Fix #3000: Set no_proxy in the okhttp builder in case the proxy_url is null

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

@@ -59,7 +59,7 @@ public Reflector(Class<T> apiTypeClass, ListerWatcher<T, L> listerWatcher, Store
     this.resyncPeriodMillis = resyncPeriodMillis;
     this.lastSyncResourceVersion = new AtomicReference<>();
     this.resyncExecutor = resyncExecutor;
-    this.watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::startWatcher, this::reListAndSync);
+    this.watcher = new ReflectorWatcher<>(store, lastSyncResourceVersion, this::stopWatcher, this::reListAndSync);
     this.isActive = new AtomicBoolean(true);
     this.isWatcherStarted = new AtomicBoolean(false);
     this.watch = new AtomicReference<>(null);
@@ -111,6 +111,15 @@ private void reListAndSync() {
     }
   }
 
+  private void stopWatcher() {
+    String ns = operationContext.getNamespace();
+    log.debug("Stopping watcher for resource {} v{} in namespace {}", apiTypeClass, lastSyncResourceVersion.get(), ns);
+    if (watch.get() != null) {
+      log.debug("Stopping watcher");
+      watch.get().close();
+    }
+  }
+
   private void startWatcher() {
     log.debug("Starting watcher for resource {} v{}", apiTypeClass, lastSyncResourceVersion.get());
     if (watch.get() != null) {

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/utils/Utils.java

@@ -163,6 +163,7 @@ public static boolean waitUntilReady(BlockingQueue<Object> queue, long amount, T
    * @return True if shutdown is complete.
    */
   public static boolean shutdownExecutorService(ExecutorService executorService) {
+    LOGGER.debug("shuting down executor: {}", executorService);
     if (executorService == null) {
       return false;
     }
@@ -173,6 +174,7 @@ public static boolean shutdownExecutorService(ExecutorService executorService) {
 
     try {
       //Wait for clean termination
+      LOGGER.debug("waiting for clean termination 5 sec");
       if (executorService.awaitTermination(5, TimeUnit.SECONDS)) {
         return true;
       }
@@ -181,20 +183,22 @@ public static boolean shutdownExecutorService(ExecutorService executorService) {
       if (!executorService.isTerminated()) {
         executorService.shutdownNow();
       }
-
+      LOGGER.debug("waiting for clean termination 5 sec");
       if (executorService.awaitTermination(5, TimeUnit.SECONDS)) {
         return true;
       }
 
       if (LOGGER.isDebugEnabled()) {
         List<Runnable> tasks = executorService.shutdownNow();
         if (!tasks.isEmpty()) {
-          LOGGER.debug("ExecutorService was not cleanly shutdown, after waiting for 10 seconds. Number of remaining tasks: {}", tasks.size());
+          LOGGER.info("ExecutorService was not cleanly shutdown, after waiting for 10 seconds. Number of remaining tasks: {}", tasks.size());
+          LOGGER.debug("Remaining tasks: {}", tasks);
         }
       }
     } catch (InterruptedException e) {
       executorService.shutdownNow();
       //Preserve interrupted status
+      LOGGER.debug("ServiceExectuor thread forcibly interrupted");
       Thread.currentThread().interrupt();
     }
     return false;