Merge pull request #1140 from yue9944882/fix/graceful-shutdown-sharedprocessor

Graceful shutdown sharedprocessor

Author: k8s-ci-robot

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

@@ -23,12 +23,16 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.commons.collections4.CollectionUtils;
 import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /*
  * SharedProcessor class manages all the registered ProcessorListener and distributes notifications.
  */
 public class SharedProcessor<ApiType extends KubernetesObject> {
 
+  private static final Logger log = LoggerFactory.getLogger(SharedProcessor.class);
+
   private ReadWriteLock lock = new ReentrantReadWriteLock();
 
   private List<ProcessorListener<ApiType>> listeners;
@@ -155,16 +159,16 @@ public void stop() {
     } finally {
       lock.writeLock().unlock();
     }
-    // Disable new tasks from being submitted
-    executorService.shutdown();
+    // Interrupts running listeners by signalling InterruptedException
+    executorService.shutdownNow();
     try {
-      // Wait a while for existing tasks to terminate
+      // Hold until all the listeners exits
       if (!executorService.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
-        // Cancel currently executing tasks
-        executorService.shutdownNow();
+        log.warn(
+            "SharedProcessors wasn't gracefully terminated, there can be listener thread leakage");
       }
     } catch (InterruptedException e) {
-      executorService.shutdownNow();
+      log.error("Graceful shutdown process of SharedProcessors was interrupted");
     }
   }
 }

util/src/test/java/io/kubernetes/client/informer/cache/SharedProcessorTest.java

@@ -67,22 +67,6 @@ public void testListenerAddition() throws InterruptedException {
   public void testShutdownGracefully() throws InterruptedException {
     SharedProcessor<V1Pod> sharedProcessor =
         new SharedProcessor<>(Executors.newCachedThreadPool(), Duration.ofSeconds(5));
-    TestWorker<V1Pod> quickWorker = new TestWorker<>(null, 0);
-    quickWorker.setTask(
-        () -> {
-          try {
-            // sleep 2s so that it could terminate within timeout(5s)
-            Thread.sleep(2000);
-          } catch (InterruptedException e) {
-          }
-        });
-    long before = System.currentTimeMillis();
-    sharedProcessor.addAndStartListener(quickWorker);
-    sharedProcessor.stop();
-    // the stopping worker properly blocks the processor's stop call
-    assertTrue(System.currentTimeMillis() - before >= 2000);
-
-    sharedProcessor = new SharedProcessor<>(Executors.newCachedThreadPool(), Duration.ofSeconds(5));
     TestWorker<V1Pod> slowWorker = new TestWorker<>(null, 0);
     final boolean[] interrupted = {false};
     slowWorker.setTask(