[CORE] Add ThreadPool.terminate to streamline shutdown

Shutting down threadpools and executor services is done in very similar
fashion across the codebase. This commit streamlines the process by
adding a terminate method to ThreadPool.

src/main/java/org/elasticsearch/client/transport/TransportClient.java

@@ -279,16 +279,8 @@ public class TransportClient extends AbstractClient {
         for (Class<? extends LifecycleComponent> plugin : pluginsService.services()) {
             injector.getInstance(plugin).close();
         }
-
-        injector.getInstance(ThreadPool.class).shutdown();
         try {
-            injector.getInstance(ThreadPool.class).awaitTermination(10, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            // ignore
-            Thread.currentThread().interrupt();
-        }
-        try {
-            injector.getInstance(ThreadPool.class).shutdownNow();
+            ThreadPool.terminate(injector.getInstance(ThreadPool.class), 10, TimeUnit.SECONDS);
         } catch (Exception e) {
             // ignore
         }

src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java

@@ -243,10 +243,8 @@ public class UnicastZenPing extends AbstractLifecycleComponent<ZenPing> implemen
 
         public void close() {
             closed = true;
-            if (executor != null) {
-                executor.shutdownNow();
-                executor = null;
-            }
+            ThreadPool.terminate(executor, 0, TimeUnit.SECONDS);
+            executor = null;
         }
     }
 

src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java

@@ -29,6 +29,7 @@ import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.node.settings.NodeSettingsService;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -95,13 +96,7 @@ public class RecoverySettings extends AbstractComponent {
     }
 
     public void close() {
-        concurrentStreamPool.shutdown();
-        try {
-            concurrentStreamPool.awaitTermination(1, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            // that's fine...
-        }
-        concurrentStreamPool.shutdownNow();
+        ThreadPool.terminate(concurrentStreamPool, 1, TimeUnit.SECONDS);
     }
 
     public ByteSizeValue fileChunkSize() {

src/main/java/org/elasticsearch/node/internal/InternalNode.java

@@ -368,6 +368,7 @@ public final class InternalNode implements Node {
         injector.getInstance(ScriptService.class).close();
 
         stopWatch.stop().start("thread_pool");
+        // TODO this should really use ThreadPool.terminate()
         injector.getInstance(ThreadPool.class).shutdown();
         try {
             injector.getInstance(ThreadPool.class).awaitTermination(10, TimeUnit.SECONDS);

src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -267,7 +267,8 @@ public class ThreadPool extends AbstractComponent {
             }
         }
         while (!retiredExecutors.isEmpty()) {
-            result &= ((ThreadPoolExecutor) retiredExecutors.remove().executor()).awaitTermination(timeout, unit);
+            ThreadPoolExecutor executor = (ThreadPoolExecutor) retiredExecutors.remove().executor();
+            result &= executor.awaitTermination(timeout, unit);
         }
         estimatedTimeThread.join(unit.toMillis(timeout));
         return result;
@@ -704,4 +705,44 @@ public class ThreadPool extends AbstractComponent {
             updateSettings(settings);
         }
     }
+
+    /**
+     * Returns <code>true</code> if the given service was terminated successfully. If the termination timed out,
+     * the service is <code>null</code> this method will return <code>false</code>.
+     */
+    public static boolean terminate(ExecutorService service, long timeout, TimeUnit timeUnit) {
+        if (service != null) {
+            service.shutdown();
+            try {
+                if (service.awaitTermination(timeout, timeUnit)) {
+                    return true;
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            service.shutdownNow();
+        }
+        return false;
+    }
+
+    /**
+     * Returns <code>true</code> if the given pool was terminated successfully. If the termination timed out,
+     * the service is <code>null</code> this method will return <code>false</code>.
+     */
+    public static boolean terminate(ThreadPool pool, long timeout, TimeUnit timeUnit) {
+        if (pool != null) {
+            pool.shutdown();
+            try {
+                if (pool.awaitTermination(timeout, timeUnit)) {
+                    return true;
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            // last resort
+            pool.shutdownNow();
+        }
+        return false;
+    }
+
 }

src/main/java/org/elasticsearch/transport/local/LocalTransport.java

@@ -117,13 +117,7 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem
 
     @Override
     protected void doClose() throws ElasticsearchException {
-        workers.shutdown();
-        try {
-            workers.awaitTermination(10, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-        }
-        workers.shutdownNow();
+        ThreadPool.terminate(workers, 10, TimeUnit.SECONDS);
     }
 
     @Override

src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java

@@ -70,14 +70,14 @@ public class TransportClientNodesServiceTests extends ElasticsearchTestCase {
         }
 
         public void close() {
-            threadPool.shutdown();
-            try {
-                threadPool.awaitTermination(1, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                Thread.currentThread().isInterrupted();
-            }
+
             transportService.stop();
             transportClientNodesService.close();
+            try {
+                terminate(threadPool);
+            } catch (InterruptedException e) {
+                throw new AssertionError(e);
+            }
         }
     }
 

src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java

@@ -153,7 +153,7 @@ public class EsExecutorsTests extends ElasticsearchTestCase {
         assertThat(executed2.get(), equalTo(true));
         assertThat(executed3.get(), equalTo(false));
 
-        executor.shutdownNow();
+        terminate(executor);
     }
 
     @Test
@@ -189,7 +189,7 @@ public class EsExecutorsTests extends ElasticsearchTestCase {
         assertThat("wrong pool size", pool.getPoolSize(), equalTo(max));
         assertThat("wrong active size", pool.getActiveCount(), equalTo(max));
         barrier.await();
-        pool.shutdown();
+        terminate(pool);
     }
 
     @Test
@@ -232,6 +232,6 @@ public class EsExecutorsTests extends ElasticsearchTestCase {
                 assertThat("idle threads didn't shrink below max. (" + pool.getPoolSize() + ")", pool.getPoolSize(), lessThan(max));
             }
         });
-        pool.shutdown();
+        terminate(pool);
     }
 }

src/test/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPingTests.java

@@ -47,7 +47,7 @@ import static org.hamcrest.Matchers.equalTo;
 public class UnicastZenPingTests extends ElasticsearchTestCase {
 
     @Test
-    public void testSimplePings() {
+    public void testSimplePings() throws InterruptedException {
         Settings settings = ImmutableSettings.EMPTY;
         int startPort = 11000 + randomIntBetween(0, 1000);
         int endPort = startPort + 10;
@@ -132,7 +132,7 @@ public class UnicastZenPingTests extends ElasticsearchTestCase {
             zenPingB.close();
             transportServiceA.close();
             transportServiceB.close();
-            threadPool.shutdown();
+            terminate(threadPool);
         }
     }
 }

src/test/java/org/elasticsearch/test/ElasticsearchTestCase.java

@@ -530,20 +530,13 @@ public abstract class ElasticsearchTestCase extends AbstractRandomizedTest {
         boolean terminated = true;
         for (ExecutorService service : services) {
             if (service != null) {
-                service.shutdown();
-                service.shutdownNow();
-                terminated &= service.awaitTermination(10, TimeUnit.SECONDS);
+                terminated &= ThreadPool.terminate(service, 10, TimeUnit.SECONDS);
             }
         }
         return terminated;
     }
 
     public static boolean terminate(ThreadPool service) throws InterruptedException {
-        if (service != null) {
-            service.shutdown();
-            service.shutdownNow();
-            return service.awaitTermination(10, TimeUnit.SECONDS);
-        }
-        return true;
+        return ThreadPool.terminate(service, 10, TimeUnit.SECONDS);
     }
 }

src/test/java/org/elasticsearch/threadpool/UpdateThreadPoolSettingsTests.java

@@ -49,7 +49,7 @@ public class UpdateThreadPoolSettingsTests extends ElasticsearchTestCase {
     }
 
     @Test
-    public void testCachedExecutorType() {
+    public void testCachedExecutorType() throws InterruptedException {
         ThreadPool threadPool = new ThreadPool(
                 ImmutableSettings.settingsBuilder()
                         .put("threadpool.search.type", "cached")
@@ -101,12 +101,11 @@ public class UpdateThreadPoolSettingsTests extends ElasticsearchTestCase {
         // Make sure executor didn't change
         assertThat(info(threadPool, Names.SEARCH).getType(), equalTo("cached"));
         assertThat(threadPool.executor(Names.SEARCH), sameInstance(oldExecutor));
-
-        threadPool.shutdown();
+        terminate(threadPool);
     }
 
     @Test
-    public void testFixedExecutorType() {
+    public void testFixedExecutorType() throws InterruptedException {
         ThreadPool threadPool = new ThreadPool(settingsBuilder()
                 .put("threadpool.search.type", "fixed")
                 .put("name","testCachedExecutorType").build(), null);
@@ -161,12 +160,12 @@ public class UpdateThreadPoolSettingsTests extends ElasticsearchTestCase {
                 .put("threadpool.search.queue", "500")
                 .build());
 
-        threadPool.shutdown();
+        terminate(threadPool);
     }
 
 
     @Test
-    public void testScalingExecutorType() {
+    public void testScalingExecutorType() throws InterruptedException {
         ThreadPool threadPool = new ThreadPool(settingsBuilder()
                 .put("threadpool.search.type", "scaling")
                 .put("threadpool.search.size", 10)
@@ -197,7 +196,7 @@ public class UpdateThreadPoolSettingsTests extends ElasticsearchTestCase {
         assertThat(((EsThreadPoolExecutor) threadPool.executor(Names.SEARCH)).getKeepAliveTime(TimeUnit.MINUTES), equalTo(10L));
         assertThat(threadPool.executor(Names.SEARCH), sameInstance(oldExecutor));
 
-        threadPool.shutdown();
+        terminate(threadPool);
     }
 
     @Test(timeout = 10000)
@@ -224,8 +223,9 @@ public class UpdateThreadPoolSettingsTests extends ElasticsearchTestCase {
         assertThat(((ThreadPoolExecutor) oldExecutor).isShutdown(), equalTo(true));
         assertThat(((ThreadPoolExecutor) oldExecutor).isTerminating(), equalTo(true));
         assertThat(((ThreadPoolExecutor) oldExecutor).isTerminated(), equalTo(false));
-        terminate(threadPool);
+        threadPool.shutdownNow(); // interrupt the thread
         latch.await();
+        terminate(threadPool);
     }
 
 }