ensuring that the reconnect task terminates

Partially addresses: fabric8io#5327

CHANGELOG.md

@@ -8,6 +8,7 @@
 * Fix #5281: Ensure the KubernetesCrudDispatcher's backing map is accessed w/lock
 * Fix #5293: Ensured the mock server uses only generic or JsonNode parsing
 * Fix #4225: [crd-generator] Principled generation of enum values instead of considering more properties
+* Fix #5327: Ensured that the informer reconnect task terminates after client close
 
 #### Improvements
 * Fix #5166: Remove opinionated messages from Config's `errorMessages` and deprecate it

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/AbstractWatchManager.java

@@ -159,7 +159,8 @@ public synchronized void closeRequest() {
     if (state != null && state.closed.compareAndSet(false, true)) {
       logger.debug("Closing the current watch");
       closeCurrentRequest();
-      CompletableFuture<Void> future = Utils.schedule(Runnable::run, () -> failSafeReconnect(state), watchEndCheckMs,
+      CompletableFuture<Void> future = Utils.schedule(baseOperation.getOperationContext().getExecutor(),
+          () -> failSafeReconnect(state), watchEndCheckMs,
           TimeUnit.MILLISECONDS);
       state.ended.whenComplete((v, t) -> future.cancel(true));
     }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/impl/DefaultSharedIndexInformer.java

@@ -88,7 +88,7 @@ public DefaultSharedIndexInformer(Class<T> apiTypeClass, ListerWatcher<T, L> lis
     this.processor = new SharedProcessor<>(informerExecutor, description);
 
     processorStore = new ProcessorStore<>(this.indexer, this.processor);
-    this.reflector = new Reflector<>(listerWatcher, processorStore);
+    this.reflector = new Reflector<>(listerWatcher, processorStore, informerExecutor);
   }
 
   /**

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

@@ -34,6 +34,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.function.LongSupplier;
@@ -54,6 +55,7 @@ public class Reflector<T extends HasMetadata, L extends KubernetesResourceList<T
   private final CompletableFuture<Void> startFuture = new CompletableFuture<>();
   private final CompletableFuture<Void> stopFuture = new CompletableFuture<>();
   private final ExponentialBackoffIntervalCalculator retryIntervalCalculator;
+  private final Executor executor;
   //default behavior - retry if started and it's not a watcherexception
   private volatile ExceptionHandler handler = (b, t) -> b && !(t instanceof WatcherException);
   private long minTimeout = MIN_TIMEOUT;
@@ -63,11 +65,16 @@ public class Reflector<T extends HasMetadata, L extends KubernetesResourceList<T
   private boolean cachedListing = true;
 
   public Reflector(ListerWatcher<T, L> listerWatcher, SyncableStore<T> store) {
+    this(listerWatcher, store, Runnable::run);
+  }
+
+  public Reflector(ListerWatcher<T, L> listerWatcher, SyncableStore<T> store, Executor executor) {
     this.listerWatcher = listerWatcher;
     this.store = store;
     this.watcher = new ReflectorWatcher();
     this.retryIntervalCalculator = new ExponentialBackoffIntervalCalculator(listerWatcher.getWatchReconnectInterval(),
         ExponentialBackoffIntervalCalculator.UNLIMITED_RETRIES);
+    this.executor = executor;
   }
 
   public CompletableFuture<Void> start() {
@@ -160,9 +167,7 @@ protected void reconnect() {
     if (isStopped()) {
       return;
     }
-    // this can be run in the scheduler thread because
-    // any further operations will happen on the io thread
-    reconnectFuture = Utils.schedule(Runnable::run, this::listSyncAndWatch,
+    reconnectFuture = Utils.schedule(executor, this::listSyncAndWatch,
         retryIntervalCalculator.nextReconnectInterval(), TimeUnit.MILLISECONDS);
   }
 
@@ -222,7 +227,7 @@ private synchronized CompletableFuture<? extends Watch> startWatcher(final Strin
       timeoutFuture.cancel(true);
     }
     timeoutFuture = new CompletableFuture<>();
-    Utils.scheduleWithVariableRate(timeoutFuture, Runnable::run,
+    Utils.scheduleWithVariableRate(timeoutFuture, executor,
         () -> future.thenAccept(AbstractWatchManager::closeRequest), timeout.getAsLong(), timeout, TimeUnit.SECONDS);
     watchFuture = future;
     return watchFuture;

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/informers/impl/cache/ReflectorTest.java

@@ -30,8 +30,12 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -134,7 +138,13 @@ void testTimeout() {
     PodList list = new PodListBuilder().withNewMetadata().withResourceVersion("1").endMetadata().build();
     Mockito.when(mock.submitList(Mockito.any())).thenReturn(CompletableFuture.completedFuture(list));
 
-    Reflector<Pod, PodList> reflector = new Reflector<>(mock, Mockito.mock(SyncableStore.class));
+    Executor ex = Mockito.mock(Executor.class);
+    Mockito.doAnswer(invocation -> {
+      ((Runnable) invocation.getArgument(0)).run();
+      return null;
+    }).when(ex).execute(Mockito.any(Runnable.class));
+
+    Reflector<Pod, PodList> reflector = new Reflector<>(mock, Mockito.mock(SyncableStore.class), ex);
     reflector.setMinTimeout(1);
 
     AbstractWatchManager manager = Mockito.mock(AbstractWatchManager.class);
@@ -154,6 +164,24 @@ void testTimeout() {
       }
       return true;
     });
+
+    // simulate an abrupt client close
+    AtomicInteger rejected = new AtomicInteger();
+    Mockito.doAnswer(invocation -> {
+      rejected.incrementAndGet();
+      throw new RejectedExecutionException();
+    }).when(ex).execute(Mockito.any(Runnable.class));
+
+    // make sure the reconnect is rejected
+    Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> {
+      return rejected.get() > 0;
+    });
+
+    long start = System.currentTimeMillis();
+    Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> {
+      assertEquals(1, rejected.get());
+      return System.currentTimeMillis() - start > 5000;
+    });
   }
 
 }