Use fluent-style builder for pool creation, replacing overloaded methods (apache#4384)

* Use fluent-style builder for pool creation, replacing overloaded methods

* Replaces overloaded createThreadPool methods with a fluent-style builder (ThreadPoolExecutorBuilderTest).
* Adds ThreadPoolExecutorBuilderTest test,
* Adds `createExecutorService` method that does not have an option to enable
metrics and replaces all occurrences in the code that was calling the alternate
method with emitThreadPoolMetrics=false to use it.

The `createScheduledExecutorService` will be refactored in a future PR when service
initialization is reworked.  See PR apache#4342 for an example.

This change isolates where metrics can be enabled and makes finding them
easier.  It also will make reviewing future changes easier because those
changes will be isolated to places that currently enable thread pool
metrics.

Author: EdColeman

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

@@ -371,7 +371,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

@@ -301,8 +301,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

@@ -495,7 +495,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

@@ -672,11 +672,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

@@ -482,12 +482,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

@@ -79,10 +79,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

@@ -222,9 +222,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,
@@ -250,9 +249,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) -> {