Merge pull request #1096 from dddddai/master

Gracefully shutdown listener pool

Author: k8s-ci-robot

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

@@ -13,10 +13,12 @@
 package io.kubernetes.client.informer.cache;
 
 import io.kubernetes.client.common.KubernetesObject;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.commons.collections4.CollectionUtils;
@@ -34,15 +36,22 @@ public class SharedProcessor<ApiType extends KubernetesObject> {
 
   private ExecutorService executorService;
 
+  private final Duration timeout;
+
   public SharedProcessor() {
     this(Executors.newCachedThreadPool());
   }
 
   public SharedProcessor(ExecutorService threadPool) {
+    this(threadPool, Duration.ofMinutes(1));
+  }
+
+  public SharedProcessor(ExecutorService threadPool, Duration timeout) {
     this.listeners = new ArrayList<>();
     this.syncingListeners = new ArrayList<>();
 
     this.executorService = threadPool;
+    this.timeout = timeout;
   }
 
   /**
@@ -146,7 +155,16 @@ public void stop() {
     } finally {
       lock.writeLock().unlock();
     }
-    // TODO(yue9944882): gracefully shutdown listener pool
-    executorService.shutdownNow();
+    // Disable new tasks from being submitted
+    executorService.shutdown();
+    try {
+      // Wait a while for existing tasks to terminate
+      if (!executorService.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
+        // Cancel currently executing tasks
+        executorService.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      executorService.shutdownNow();
+    }
   }
 }

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

@@ -18,6 +18,8 @@
 import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.openapi.models.V1Pod;
+import java.time.Duration;
+import java.util.concurrent.Executors;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.EnvironmentVariables;
@@ -61,6 +63,45 @@ public void testListenerAddition() throws InterruptedException {
     assertTrue(expectDeleteHandler.isSatisfied());
   }
 
+  @Test
+  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(
+        () -> {
+          try {
+            // sleep 10s so that it could be interrupted by shutdownNow()
+            Thread.sleep(10 * 1000);
+          } catch (InterruptedException e) {
+            interrupted[0] = true;
+          }
+        });
+    sharedProcessor.addAndStartListener(slowWorker);
+    sharedProcessor.stop();
+    // make sure the slow worker has set interrupted[0] = true
+    Thread.sleep(1000);
+    // the slow worker is interrupted upon timeout
+    assertTrue(interrupted[0]);
+  }
+
   private static class ExpectingNoticationHandler<ApiType extends KubernetesObject>
       extends ProcessorListener<ApiType> {
 
@@ -99,4 +140,23 @@ public boolean isSatisfied() {
       return satisfied;
     }
   }
+
+  private static class TestWorker<ApiType extends KubernetesObject>
+      extends ProcessorListener<ApiType> {
+
+    private Runnable task;
+
+    public TestWorker(ResourceEventHandler<ApiType> handler, long resyncPeriod) {
+      super(handler, resyncPeriod);
+    }
+
+    public void setTask(Runnable task) {
+      this.task = task;
+    }
+
+    @Override
+    public void run() {
+      this.task.run();
+    }
+  }
 }