Merge remote-tracking branch 'upstream/2.1'

Author: Ed Coleman

core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java

@@ -260,17 +260,18 @@ public Ample getAmple() {
       submitScannerReadAheadTask(Callable<List<KeyValue>> c) {
     ensureOpen();
     if (scannerReadaheadPool == null) {
-      scannerReadaheadPool = clientThreadPools.createThreadPool(0, Integer.MAX_VALUE, 3L, SECONDS,
-          "Accumulo scanner read ahead thread", new SynchronousQueue<>(), true);
+      scannerReadaheadPool = clientThreadPools.getPoolBuilder("Accumulo scanner read ahead thread")
+          .numCoreThreads(0).numMaxThreads(Integer.MAX_VALUE).withTimeOut(3L, SECONDS)
+          .withQueue(new SynchronousQueue<>()).enableThreadPoolMetrics().build();
     }
     return scannerReadaheadPool.submit(c);
   }
 
   public synchronized void executeCleanupTask(Runnable r) {
     ensureOpen();
     if (cleanupThreadPool == null) {
-      cleanupThreadPool = clientThreadPools.createFixedThreadPool(1, 3, SECONDS,
-          "Conditional Writer Cleanup Thread", true);
+      cleanupThreadPool = clientThreadPools.getPoolBuilder("Conditional Writer Cleanup Thread")
+          .numCoreThreads(1).withTimeOut(3L, SECONDS).enableThreadPoolMetrics().build();
     }
     this.cleanupThreadPool.execute(r);
   }

core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java

@@ -372,7 +372,7 @@ private TabletServerMutations<QCMutation> dequeue(String location) {
     this.auths = config.getAuthorizations();
     this.ve = new VisibilityEvaluator(config.getAuthorizations());
     this.threadPool = context.threadPools().createScheduledExecutorService(
-        config.getMaxWriteThreads(), this.getClass().getSimpleName(), false);
+        config.getMaxWriteThreads(), this.getClass().getSimpleName());
     this.locator = new SyncingTabletLocator(context, tableId);
     this.serverQueues = new HashMap<>();
     this.tableId = tableId;

core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java

@@ -307,8 +307,8 @@ public List<ActiveCompaction> getActiveCompactions()
     List<String> tservers = getTabletServers();
 
     int numThreads = Math.max(4, Math.min((tservers.size() + compactors.size()) / 10, 256));
-    var executorService =
-        context.threadPools().createFixedThreadPool(numThreads, "getactivecompactions", false);
+    var executorService = context.threadPools().getPoolBuilder("getactivecompactions")
+        .numCoreThreads(numThreads).build();
     try {
       List<Future<List<ActiveCompaction>>> futures = new ArrayList<>();
 

core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java

@@ -493,7 +493,8 @@ public void addSplits(String tableName, SortedSet<Text> partitionKeys)
     CountDownLatch latch = new CountDownLatch(splits.size());
     AtomicReference<Exception> exception = new AtomicReference<>(null);
 
-    ExecutorService executor = context.threadPools().createFixedThreadPool(16, "addSplits", false);
+    ExecutorService executor =
+        context.threadPools().getPoolBuilder("addSplits").numCoreThreads(16).build();
     try {
       executor.execute(
           new SplitTask(new SplitEnv(tableName, tableId, executor, latch, exception), splits));

core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReader.java

@@ -71,8 +71,9 @@ protected TabletServerBatchReader(ClientContext context, Class<?> scopeClass, Ta
     this.tableName = tableName;
     this.numThreads = numQueryThreads;
 
-    queryThreadPool = context.threadPools().createFixedThreadPool(numQueryThreads,
-        "batch scanner " + batchReaderInstance + "-", false);
+    queryThreadPool =
+        context.threadPools().getPoolBuilder("batch scanner " + batchReaderInstance + "-")
+            .numCoreThreads(numQueryThreads).build();
     // Call shutdown on this thread pool in case the caller does not call close().
     cleanable = CleanerUtil.shutdownThreadPoolExecutor(queryThreadPool, closed, log);
   }

core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchWriter.java

@@ -671,11 +671,11 @@ private class MutationWriter {
     public MutationWriter(int numSendThreads) {
       serversMutations = new HashMap<>();
       queued = new HashSet<>();
-      sendThreadPool = context.threadPools().createFixedThreadPool(numSendThreads,
-          this.getClass().getName(), false);
+      sendThreadPool = context.threadPools().getPoolBuilder(this.getClass().getName())
+          .numCoreThreads(numSendThreads).build();
       locators = new HashMap<>();
-      binningThreadPool = context.threadPools().createFixedThreadPool(1, "BinMutations",
-          new SynchronousQueue<>(), false);
+      binningThreadPool = context.threadPools().getPoolBuilder("BinMutations").numCoreThreads(1)
+          .withQueue(new SynchronousQueue<>()).build();
       binningThreadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
     }
 

core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java

@@ -464,12 +464,12 @@ private SortedMap<KeyExtent,Bulk.Files> computeMappingFromFiles(FileSystem fs, T
     if (this.executor != null) {
       executor = this.executor;
     } else if (numThreads > 0) {
-      executor = service =
-          context.threadPools().createFixedThreadPool(numThreads, "BulkImportThread", false);
+      executor = service = context.threadPools().getPoolBuilder("BulkImportThread")
+          .numCoreThreads(numThreads).build();
     } else {
       String threads = context.getConfiguration().get(ClientProperty.BULK_LOAD_THREADS.getKey());
-      executor = service = context.threadPools().createFixedThreadPool(
-          ConfigurationTypeHelper.getNumThreads(threads), "BulkImportThread", false);
+      executor = service = context.threadPools().getPoolBuilder("BulkImportThread")
+          .numCoreThreads(ConfigurationTypeHelper.getNumThreads(threads)).build();
     }
 
     try {

core/src/main/java/org/apache/accumulo/core/file/BloomFilterLayer.java

@@ -80,10 +80,9 @@ private static synchronized ExecutorService getLoadThreadPool(int maxLoadThreads
     }
 
     if (maxLoadThreads > 0) {
-      loadThreadPool = ThreadPools.getServerThreadPools().createThreadPool(0, maxLoadThreads, 60,
-          SECONDS, "bloom-loader", false);
+      loadThreadPool = ThreadPools.getServerThreadPools().getPoolBuilder("bloom-loader")
+          .numCoreThreads(0).numMaxThreads(maxLoadThreads).withTimeOut(60L, SECONDS).build();
     }
-
     return loadThreadPool;
   }
 

core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCache.java

@@ -102,8 +102,8 @@ public class LruBlockCache extends SynchronousLoadingBlockCache implements Block
   private final EvictionThread evictionThread;
 
   /** Statistics thread schedule pool (for heavy debugging, could remove) */
-  private final ScheduledExecutorService scheduleThreadPool = ThreadPools.getServerThreadPools()
-      .createScheduledExecutorService(1, "LRUBlockCacheStats", false);
+  private final ScheduledExecutorService scheduleThreadPool =
+      ThreadPools.getServerThreadPools().createScheduledExecutorService(1, "LRUBlockCacheStats");
 
   /** Current size of cache */
   private final AtomicLong size;

core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/tinylfu/TinyLfuBlockCache.java

@@ -62,7 +62,7 @@ public final class TinyLfuBlockCache implements BlockCache {
   private final Policy.Eviction<String,Block> policy;
   private final int maxSize;
   private final ScheduledExecutorService statsExecutor = ThreadPools.getServerThreadPools()
-      .createScheduledExecutorService(1, "TinyLfuBlockCacheStatsExecutor", false);
+      .createScheduledExecutorService(1, "TinyLfuBlockCacheStatsExecutor");
 
   public TinyLfuBlockCache(Configuration conf, CacheType type) {
     cache = Caffeine.newBuilder()

core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java

@@ -218,9 +218,8 @@ private static ExternalCompactionId getRunningCompactionId(HostAndPort compactor
    */
   public static List<RunningCompaction> getCompactionsRunningOnCompactors(ClientContext context) {
     final List<RunningCompactionFuture> rcFutures = new ArrayList<>();
-    final ExecutorService executor = ThreadPools.getServerThreadPools().createFixedThreadPool(16,
-        "CompactorRunningCompactions", false);
-
+    final ExecutorService executor = ThreadPools.getServerThreadPools()
+        .getPoolBuilder("CompactorRunningCompactions").numCoreThreads(16).build();
     getCompactorAddrs(context).forEach((q, hp) -> {
       hp.forEach(hostAndPort -> {
         rcFutures.add(new RunningCompactionFuture(q, hostAndPort,
@@ -246,9 +245,8 @@ public static List<RunningCompaction> getCompactionsRunningOnCompactors(ClientCo
 
   public static Collection<ExternalCompactionId>
       getCompactionIdsRunningOnCompactors(ClientContext context) {
-    final ExecutorService executor = ThreadPools.getServerThreadPools().createFixedThreadPool(16,
-        "CompactorRunningCompactions", false);
-
+    final ExecutorService executor = ThreadPools.getServerThreadPools()
+        .getPoolBuilder("CompactorRunningCompactions").numCoreThreads(16).build();
     List<Future<ExternalCompactionId>> futures = new ArrayList<>();
 
     getCompactorAddrs(context).forEach((q, hp) -> {