shut down reflector daemon job gracefully

Author: yue9944882

util/src/main/java/io/kubernetes/client/informer/cache/Controller.java

@@ -96,28 +96,34 @@ public void run() {
       log.info("informer#Controller: resync skipped due to 0 full resync period");
     }
 
-    // TODO(yue9944882): proper naming for reflector
-    reflector = new ReflectorRunnable<ApiType, ApiListType>(apiTypeClass, listerWatcher, queue);
-
-    reflectorFuture =
-        reflectExecutor.scheduleWithFixedDelay(
-            reflector::run, 0L, DEFAULT_PERIOD, TimeUnit.MILLISECONDS);
+    synchronized (this) {
+      // TODO(yue9944882): proper naming for reflector
+      reflector = new ReflectorRunnable<ApiType, ApiListType>(apiTypeClass, listerWatcher, queue);
+      try {
+        reflectorFuture =
+            reflectExecutor.scheduleWithFixedDelay(
+                reflector::run, 0L, DEFAULT_PERIOD, TimeUnit.MILLISECONDS);
+      } catch (RejectedExecutionException e) {
+        // submitting reflector list-watching job can fail due to concurrent invocation of
+        // `shutdown`. handling exception with a warning then return.
+        log.warn("reflector list-watching job exiting because the thread-pool is shutting down");
+        return;
+      }
+    }
 
     // start the process loop
     this.processLoop();
   }
 
   /** stops the resync thread pool firstly, then stop the reflector */
   public void stop() {
-
-    reflector.stop();
-    reflectorFuture.cancel(true);
-    reflectExecutor.shutdown();
-
-    if (resyncFuture != null) {
-      resyncFuture.cancel(true);
-      resyncExecutor.shutdown();
+    synchronized (this) {
+      if (reflectorFuture != null) {
+        reflector.stop();
+        reflectorFuture.cancel(true);
+      }
     }
+    reflectExecutor.shutdown();
   }
 
   /** returns true if the queue has been resycned */

util/src/test/java/io/kubernetes/client/informer/SharedInformerFactoryTest.java

@@ -0,0 +1,40 @@
+package io.kubernetes.client.informer;
+
+import io.kubernetes.client.ApiException;
+import io.kubernetes.client.apis.CoreV1Api;
+import io.kubernetes.client.models.V1Namespace;
+import io.kubernetes.client.models.V1NamespaceList;
+import io.kubernetes.client.util.CallGeneratorParams;
+import org.junit.Test;
+
+public class SharedInformerFactoryTest {
+  @Test
+  public void shutdownInformerFactoryInstantlyAfterStarting() {
+    CoreV1Api api = new CoreV1Api();
+    SharedInformerFactory factory = new SharedInformerFactory();
+    factory.sharedIndexInformerFor(
+        (CallGeneratorParams params) -> {
+          try {
+            return api.listNamespaceCall(
+                null,
+                null,
+                null,
+                null,
+                null,
+                null,
+                params.resourceVersion,
+                params.timeoutSeconds,
+                params.watch,
+                null,
+                null);
+          } catch (ApiException e) {
+            throw new RuntimeException(e);
+          }
+        },
+        V1Namespace.class,
+        V1NamespaceList.class);
+
+    factory.startAllRegisteredInformers();
+    factory.stopAllRegisteredInformers();
+  }
+}