[TEST] Add a more restrictive thread leaks filter

Today all threads are allowed to leak a suite. This is tricky since
it essentially allows resource leaks by default where for instance
test private TransportClients will never get closed and consume
resources influencing other tests. It also hides threads that
are not fully under elasticsearchs control like the Lucene
TimeLimitingCollector thread. This commit restricts the threads
that can leak a suite to the threads spawned from testclusters
and fixes sevearl places that leaked threads.

Closes #7833

core-signatures.txt

@@ -111,3 +111,7 @@ com.ning.compress.lzf.LZFOutputStream#<init>(java.io.OutputStream)
 com.ning.compress.lzf.LZFOutputStream#<init>(java.io.OutputStream, com.ning.compress.BufferRecycler)
 com.ning.compress.lzf.LZFUncompressor#<init>(com.ning.compress.DataHandler)
 com.ning.compress.lzf.LZFUncompressor#<init>(com.ning.compress.DataHandler, com.ning.compress.BufferRecycler)
+
+@defaultMessage Spawns a new thread which is solely under lucenes control use ThreadPool#estimatedTimeInMillisCounter instead
+org.apache.lucene.search.TimeLimitingCollector#getGlobalTimerThread()
+org.apache.lucene.search.TimeLimitingCollector#getGlobalCounter()

src/main/java/org/elasticsearch/action/admin/indices/validate/query/TransportValidateQueryAction.java

@@ -178,7 +178,7 @@ protected ShardValidateQueryResponse shardOperation(ShardValidateQueryRequest re
                 new ShardSearchRequest(request).types(request.types()).nowInMillis(request.nowInMillis())
                         .filteringAliases(request.filteringAliases()),
                 null, indexShard.acquireSearcher("validate_query"), indexService, indexShard,
-                scriptService, pageCacheRecycler, bigArrays
+                scriptService, pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter()
         );
         SearchContext.setCurrent(searchContext);
         try {

src/main/java/org/elasticsearch/action/count/TransportCountAction.java

@@ -170,7 +170,7 @@ protected ShardCountResponse shardOperation(ShardCountRequest request) throws El
                         .filteringAliases(request.filteringAliases())
                         .nowInMillis(request.nowInMillis()),
                 shardTarget, indexShard.acquireSearcher("count"), indexService, indexShard,
-                scriptService, pageCacheRecycler, bigArrays);
+                scriptService, pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter());
         SearchContext.setCurrent(context);
 
         try {

src/main/java/org/elasticsearch/action/deletebyquery/TransportShardDeleteByQueryAction.java

@@ -107,7 +107,7 @@ protected PrimaryResponse<ShardDeleteByQueryResponse, ShardDeleteByQueryRequest>
 
         SearchContext.setCurrent(new DefaultSearchContext(0, new ShardSearchRequest(shardRequest.request).types(request.types()).nowInMillis(request.nowInMillis()), null,
                 indexShard.acquireSearcher(DELETE_BY_QUERY_API), indexService, indexShard, scriptService,
-                pageCacheRecycler, bigArrays));
+                pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter()));
         try {
             Engine.DeleteByQuery deleteByQuery = indexShard.prepareDeleteByQuery(request.source(), request.filteringAliases(), Engine.Operation.Origin.PRIMARY, request.types());
             SearchContext.current().parsedQuery(new ParsedQuery(deleteByQuery.query(), ImmutableMap.<String, Filter>of()));
@@ -129,7 +129,7 @@ protected void shardOperationOnReplica(ReplicaOperationRequest shardRequest) {
 
         SearchContext.setCurrent(new DefaultSearchContext(0, new ShardSearchRequest(shardRequest).types(request.types()).nowInMillis(request.nowInMillis()), null,
                 indexShard.acquireSearcher(DELETE_BY_QUERY_API, IndexShard.Mode.WRITE), indexService, indexShard, scriptService,
-                pageCacheRecycler, bigArrays));
+                pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter()));
         try {
             Engine.DeleteByQuery deleteByQuery = indexShard.prepareDeleteByQuery(request.source(), request.filteringAliases(), Engine.Operation.Origin.REPLICA, request.types());
             SearchContext.current().parsedQuery(new ParsedQuery(deleteByQuery.query(), ImmutableMap.<String, Filter>of()));

src/main/java/org/elasticsearch/action/exists/TransportExistsAction.java

@@ -171,7 +171,7 @@ protected ShardExistsResponse shardOperation(ShardExistsRequest request) throws
                         .filteringAliases(request.filteringAliases())
                         .nowInMillis(request.nowInMillis()),
                 shardTarget, indexShard.acquireSearcher("exists"), indexService, indexShard,
-                scriptService, pageCacheRecycler, bigArrays);
+                scriptService, pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter());
         SearchContext.setCurrent(context);
 
         try {

src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java

@@ -119,7 +119,7 @@ protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId
                         .nowInMillis(request.nowInMillis),
                 null, result.searcher(), indexService, indexShard,
                 scriptService, pageCacheRecycler,
-                bigArrays
+                bigArrays, threadPool.estimatedTimeInMillisCounter()
         );
         SearchContext.setCurrent(context);
 

src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -28,6 +28,7 @@
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.Counter;
 import org.apache.lucene.util.Version;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchIllegalArgumentException;
@@ -149,8 +150,8 @@ public final static EarlyTerminatingCollector wrapCountBasedEarlyTerminatingColl
     /**
      * Wraps <code>delegate</code> with a time limited collector with a timeout of <code>timeoutInMillis</code>
      */
-    public final static TimeLimitingCollector wrapTimeLimitingCollector(final Collector delegate, long timeoutInMillis) {
-        return new TimeLimitingCollector(delegate, TimeLimitingCollector.getGlobalCounter(), timeoutInMillis);
+    public final static TimeLimitingCollector wrapTimeLimitingCollector(final Collector delegate, final Counter counter, long timeoutInMillis) {
+        return new TimeLimitingCollector(delegate, counter, timeoutInMillis);
     }
 
     /**

src/main/java/org/elasticsearch/percolator/PercolateContext.java

@@ -24,6 +24,7 @@
 import org.apache.lucene.index.IndexableField;
 import org.apache.lucene.search.*;
 import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.Counter;
 import org.elasticsearch.action.percolate.PercolateShardRequest;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.cache.recycler.PageCacheRecycler;
@@ -687,4 +688,9 @@ public boolean useSlowScroll() {
     public SearchContext useSlowScroll(boolean useSlowScroll) {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public Counter timeEstimateCounter() {
+        throw new UnsupportedOperationException();
+    }
 }

src/main/java/org/elasticsearch/search/SearchService.java

@@ -523,7 +523,7 @@ final SearchContext createContext(ShardSearchRequest request, @Nullable Engine.S
         SearchShardTarget shardTarget = new SearchShardTarget(clusterService.localNode().id(), request.index(), request.shardId());
 
         Engine.Searcher engineSearcher = searcher == null ? indexShard.acquireSearcher("search") : searcher;
-        SearchContext context = new DefaultSearchContext(idGenerator.incrementAndGet(), request, shardTarget, engineSearcher, indexService, indexShard, scriptService, pageCacheRecycler, bigArrays);
+        SearchContext context = new DefaultSearchContext(idGenerator.incrementAndGet(), request, shardTarget, engineSearcher, indexService, indexShard, scriptService, pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter());
         SearchContext.setCurrent(context);
         try {
             context.scroll(request.scroll());

src/main/java/org/elasticsearch/search/aggregations/metrics/tophits/TopHitsContext.java

@@ -24,6 +24,7 @@
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.ScoreDoc;
 import org.apache.lucene.search.Sort;
+import org.apache.lucene.util.Counter;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.cache.recycler.PageCacheRecycler;
 import org.elasticsearch.common.util.BigArrays;
@@ -609,4 +610,9 @@ public boolean useSlowScroll() {
     public SearchContext useSlowScroll(boolean useSlowScroll) {
         throw new UnsupportedOperationException("Not supported");
     }
+
+    @Override
+    public Counter timeEstimateCounter() {
+        throw new UnsupportedOperationException("Not supported");
+    }
 }

src/main/java/org/elasticsearch/search/internal/ContextIndexSearcher.java

@@ -132,7 +132,7 @@ public void search(List<AtomicReaderContext> leaves, Weight weight, Collector co
         if (timeoutSet) {
             // TODO: change to use our own counter that uses the scheduler in ThreadPool
             // throws TimeLimitingCollector.TimeExceededException when timeout has reached
-            collector = Lucene.wrapTimeLimitingCollector(collector, searchContext.timeoutInMillis());
+            collector = Lucene.wrapTimeLimitingCollector(collector, searchContext.timeEstimateCounter(), searchContext.timeoutInMillis());
         }
         if (terminateAfterSet) {
             // throws Lucene.EarlyTerminationException when given count is reached

src/main/java/org/elasticsearch/search/internal/DefaultSearchContext.java

@@ -25,6 +25,7 @@
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.ScoreDoc;
 import org.apache.lucene.search.Sort;
+import org.apache.lucene.util.Counter;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.cache.recycler.PageCacheRecycler;
@@ -83,6 +84,7 @@ public class DefaultSearchContext extends SearchContext {
     private final ShardSearchRequest request;
 
     private final SearchShardTarget shardTarget;
+    private final Counter timeEstimateCounter;
 
     private SearchType searchType;
 
@@ -179,7 +181,7 @@ public class DefaultSearchContext extends SearchContext {
     public DefaultSearchContext(long id, ShardSearchRequest request, SearchShardTarget shardTarget,
                          Engine.Searcher engineSearcher, IndexService indexService, IndexShard indexShard,
                          ScriptService scriptService, PageCacheRecycler pageCacheRecycler,
-                         BigArrays bigArrays) {
+                         BigArrays bigArrays, Counter timeEstimateCounter) {
         this.id = id;
         this.request = request;
         this.searchType = request.searchType();
@@ -199,6 +201,7 @@ public DefaultSearchContext(long id, ShardSearchRequest request, SearchShardTarg
 
         // initialize the filtering alias based on the provided filters
         aliasFilter = indexService.aliasesService().aliasFilter(request.filteringAliases());
+        this.timeEstimateCounter = timeEstimateCounter;
     }
 
     @Override
@@ -705,4 +708,9 @@ public DefaultSearchContext useSlowScroll(boolean useSlowScroll) {
         this.useSlowScroll = useSlowScroll;
         return this;
     }
+
+    @Override
+    public Counter timeEstimateCounter() {
+        return timeEstimateCounter;
+    }
 }

src/main/java/org/elasticsearch/search/internal/SearchContext.java

@@ -25,6 +25,7 @@
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.ScoreDoc;
 import org.apache.lucene.search.Sort;
+import org.apache.lucene.util.Counter;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.cache.recycler.PageCacheRecycler;
 import org.elasticsearch.common.Nullable;
@@ -354,6 +355,8 @@ public void clearReleasables(Lifetime lifetime) {
 
     public abstract SearchContext useSlowScroll(boolean useSlowScroll);
 
+    public abstract Counter timeEstimateCounter();
+
     /**
      * The life time of an object that is used during search execution.
      */

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

@@ -23,6 +23,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.lucene.util.Counter;
 import org.elasticsearch.ElasticsearchIllegalArgumentException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
@@ -150,6 +151,10 @@ public long estimatedTimeInMillis() {
         return estimatedTimeThread.estimatedTimeInMillis();
     }
 
+    public Counter estimatedTimeInMillisCounter() {
+        return estimatedTimeThread.counter;
+    }
+
     public ThreadPoolInfo info() {
         List<Info> infos = new ArrayList<>();
         for (ExecutorHolder holder : executors.values()) {
@@ -264,6 +269,7 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE
         while (!retiredExecutors.isEmpty()) {
             result &= ((ThreadPoolExecutor) retiredExecutors.remove().executor()).awaitTermination(timeout, unit);
         }
+        estimatedTimeThread.join(unit.toMillis(timeout));
         return result;
     }
 
@@ -506,15 +512,15 @@ public String toString() {
     static class EstimatedTimeThread extends Thread {
 
         final long interval;
-
+        final TimeCounter counter;
         volatile boolean running = true;
-
         volatile long estimatedTimeInMillis;
 
         EstimatedTimeThread(String name, long interval) {
             super(name);
             this.interval = interval;
             this.estimatedTimeInMillis = System.currentTimeMillis();
+            this.counter = new TimeCounter();
             setDaemon(true);
         }
 
@@ -534,6 +540,19 @@ public void run() {
                 }
             }
         }
+
+        private class TimeCounter extends Counter {
+
+            @Override
+            public long addAndGet(long delta) {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public long get() {
+                return estimatedTimeInMillis;
+            }
+        }
     }
 
     static class ExecutorHolder {

src/test/java/org/elasticsearch/client/node/NodeClientHeadersTests.java

@@ -50,8 +50,8 @@ public void init() {
     }
 
     @After
-    public void cleanup() {
-        threadPool.shutdownNow();
+    public void cleanup() throws InterruptedException {
+        terminate(threadPool);
     }
 
     @Override

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

@@ -94,8 +94,9 @@ public void testThatTransportClientSettingIsSet() {
 
     @Test
     public void testThatTransportClientSettingCannotBeChanged() {
-        TransportClient client = new TransportClient(settingsBuilder().put(Client.CLIENT_TYPE_SETTING, "anything"));
-        Settings settings = client.injector.getInstance(Settings.class);
-        assertThat(settings.get(Client.CLIENT_TYPE_SETTING), is("transport"));
+        try (TransportClient client = new TransportClient(settingsBuilder().put(Client.CLIENT_TYPE_SETTING, "anything"))) {
+            Settings settings = client.injector.getInstance(Settings.class);
+            assertThat(settings.get(Client.CLIENT_TYPE_SETTING), is("transport"));
+        }
     }
 }

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

@@ -60,7 +60,7 @@ public void testPriorityQueue() throws Exception {
 
     @Test
     public void testSubmitPrioritizedExecutorWithRunnables() throws Exception {
-        ExecutorService executor = EsExecutors.newSinglePrioritizing(Executors.defaultThreadFactory());
+        ExecutorService executor = EsExecutors.newSinglePrioritizing(EsExecutors.daemonThreadFactory(getTestName()));
         List<Integer> results = new ArrayList<>(8);
         CountDownLatch awaitingLatch = new CountDownLatch(1);
         CountDownLatch finishedLatch = new CountDownLatch(8);
@@ -85,11 +85,12 @@ public void testSubmitPrioritizedExecutorWithRunnables() throws Exception {
         assertThat(results.get(5), equalTo(5));
         assertThat(results.get(6), equalTo(6));
         assertThat(results.get(7), equalTo(7));
+        terminate(executor);
     }
 
     @Test
     public void testExecutePrioritizedExecutorWithRunnables() throws Exception {
-        ExecutorService executor = EsExecutors.newSinglePrioritizing(Executors.defaultThreadFactory());
+        ExecutorService executor = EsExecutors.newSinglePrioritizing(EsExecutors.daemonThreadFactory(getTestName()));
         List<Integer> results = new ArrayList<>(8);
         CountDownLatch awaitingLatch = new CountDownLatch(1);
         CountDownLatch finishedLatch = new CountDownLatch(8);
@@ -114,11 +115,12 @@ public void testExecutePrioritizedExecutorWithRunnables() throws Exception {
         assertThat(results.get(5), equalTo(5));
         assertThat(results.get(6), equalTo(6));
         assertThat(results.get(7), equalTo(7));
+        terminate(executor);
     }
 
     @Test
     public void testSubmitPrioritizedExecutorWithCallables() throws Exception {
-        ExecutorService executor = EsExecutors.newSinglePrioritizing(Executors.defaultThreadFactory());
+        ExecutorService executor = EsExecutors.newSinglePrioritizing(EsExecutors.daemonThreadFactory(getTestName()));
         List<Integer> results = new ArrayList<>(8);
         CountDownLatch awaitingLatch = new CountDownLatch(1);
         CountDownLatch finishedLatch = new CountDownLatch(8);
@@ -143,11 +145,12 @@ public void testSubmitPrioritizedExecutorWithCallables() throws Exception {
         assertThat(results.get(5), equalTo(5));
         assertThat(results.get(6), equalTo(6));
         assertThat(results.get(7), equalTo(7));
+        terminate(executor);
     }
 
     @Test
     public void testSubmitPrioritizedExecutorWithMixed() throws Exception {
-        ExecutorService executor = EsExecutors.newSinglePrioritizing(Executors.defaultThreadFactory());
+        ExecutorService executor = EsExecutors.newSinglePrioritizing(EsExecutors.daemonThreadFactory(getTestName()));
         List<Integer> results = new ArrayList<>(8);
         CountDownLatch awaitingLatch = new CountDownLatch(1);
         CountDownLatch finishedLatch = new CountDownLatch(8);
@@ -172,12 +175,13 @@ public void testSubmitPrioritizedExecutorWithMixed() throws Exception {
         assertThat(results.get(5), equalTo(5));
         assertThat(results.get(6), equalTo(6));
         assertThat(results.get(7), equalTo(7));
+        terminate(executor);
     }
 
     @Test
     public void testTimeout() throws Exception {
-        ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
-        PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(Executors.defaultThreadFactory());
+        ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(EsExecutors.daemonThreadFactory(getTestName()));
+        PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(EsExecutors.daemonThreadFactory(getTestName()));
         final CountDownLatch invoked = new CountDownLatch(1);
         final CountDownLatch block = new CountDownLatch(1);
         executor.execute(new Runnable() {
@@ -233,9 +237,7 @@ public void run() {
         block.countDown();
         Thread.sleep(100); // sleep a bit to double check that execute on the timed out update task is not called...
         assertThat(executeCalled.get(), equalTo(false));
-
-        timer.shutdownNow();
-        executor.shutdownNow();
+        assertTrue(terminate(timer, executor));
     }
 
     static class AwaitingJob extends PrioritizedRunnable {

src/test/java/org/elasticsearch/index/engine/internal/InternalEngineTests.java

@@ -146,10 +146,7 @@ public void tearDown() throws Exception {
 
         engine.close();
         store.close();
-
-        if (threadPool != null) {
-            threadPool.shutdownNow();
-        }
+        terminate(threadPool);
     }
 
     private Document testDocumentWithTextField() {

src/test/java/org/elasticsearch/index/query/TemplateQueryParserTest.java

@@ -46,7 +46,9 @@
 import org.elasticsearch.indices.query.IndicesQueriesModule;
 import org.elasticsearch.script.ScriptModule;
 import org.elasticsearch.test.ElasticsearchTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.threadpool.ThreadPoolModule;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -97,6 +99,12 @@ protected void configure() {
         context = new QueryParseContext(index, queryParserService);
     }
 
+    @After
+    public void tearDown() throws Exception {
+        super.tearDown();
+        terminate(injector.getInstance(ThreadPool.class));
+    }
+
     @Test
     public void testParser() throws IOException {
         String templateString = "{\"template\": {"