Executors/Threads: Enforce reasonable rules

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -425,10 +425,10 @@ public CoreContainer(NodeConfig config, CoresLocator locator, boolean asyncSolrC
     this.replayUpdatesExecutor =
         new OrderedExecutor<>(
             cfg.getReplayUpdatesThreads(),
-            ExecutorUtil.newMDCAwareCachedThreadPool(
+            ExecutorUtil.newMDCAwareFixedThreadPool(
                 cfg.getReplayUpdatesThreads(), // thread count
                 cfg.getReplayUpdatesThreads(), // queue size
-                new SolrNamedThreadFactory("replayUpdatesExecutor")));
+                "replayUpdatesExecutor"));
     this.appHandlersByConfigSetId = new JerseyAppHandlerCache();
 
     SolrPaths.AllowPathBuilder allowPathBuilder = new SolrPaths.AllowPathBuilder();
@@ -451,10 +451,8 @@ public CoreContainer(NodeConfig config, CoresLocator locator, boolean asyncSolrC
     this.indexSearcherExecutor = SolrIndexSearcher.initCollectorExecutor(cfg);
 
     this.indexFingerprintExecutor =
-        ExecutorUtil.newMDCAwareCachedThreadPool(
-            EXECUTOR_MAX_CPU_THREADS,
-            Integer.MAX_VALUE,
-            new SolrNamedThreadFactory("IndexFingerprintPool"));
+        ExecutorUtil.newMDCAwareFixedThreadPool(
+            EXECUTOR_MAX_CPU_THREADS, Integer.MAX_VALUE, "IndexFingerprintPool");
   }
 
   @SuppressWarnings({"unchecked"})

solr/core/src/java/org/apache/solr/core/SolrCores.java

@@ -109,7 +109,7 @@ public void close() {
 
       ExecutorService coreCloseExecutor =
           ExecutorUtil.newMDCAwareFixedThreadPool(
-              Integer.MAX_VALUE, new SolrNamedThreadFactory("coreCloseExecutor"));
+              64, new SolrNamedThreadFactory("coreCloseExecutor"));
       try {
         for (SolrCore core : coreList) {
           coreCloseExecutor.execute(

solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java

@@ -41,7 +41,6 @@
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.util.EnvUtils;
 import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.core.DirectoryFactory;
 import org.apache.solr.core.IndexDeletionPolicyWrapper;
 import org.apache.solr.core.SolrCore;
@@ -218,10 +217,8 @@ private BackupStats incrementalCopy(Collection<String> indexFiles, Directory dir
                 "BackupUploadExecutor",
                 ExecutorService.class,
                 s ->
-                    ExecutorUtil.newMDCAwareCachedThreadPool(
-                        MAX_PARALLEL_UPLOADS,
-                        Integer.MAX_VALUE,
-                        new SolrNamedThreadFactory("BackupUploadExecutor")));
+                    ExecutorUtil.newMDCAwareFixedThreadPool(
+                        MAX_PARALLEL_UPLOADS, Integer.MAX_VALUE, "BackupUploadExecutor"));
 
     List<Future<?>> uploadFutures = new ArrayList<>();
     for (String fileName : indexFiles) {

solr/core/src/java/org/apache/solr/handler/RestoreCore.java

@@ -40,7 +40,6 @@
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.util.EnvUtils;
 import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.core.DirectoryFactory;
 import org.apache.solr.core.SolrCore;
 import org.apache.solr.core.backup.BackupFilePaths;
@@ -136,10 +135,8 @@ public boolean doRestore() throws Exception {
                   "RestoreDownloadExecutor",
                   ExecutorService.class,
                   s ->
-                      ExecutorUtil.newMDCAwareCachedThreadPool(
-                          MAX_PARALLEL_DOWNLOADS,
-                          Integer.MAX_VALUE,
-                          new SolrNamedThreadFactory("RestoreDownloadExecutor")));
+                      ExecutorUtil.newMDCAwareFixedThreadPool(
+                          MAX_PARALLEL_DOWNLOADS, Integer.MAX_VALUE, "RestoreDownloadExecutor"));
 
       // Move all files from backupDir to restoreIndexDir
       for (String filename : repository.listAllFiles()) {

solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java

@@ -383,10 +383,8 @@ public static class CoreAdminAsyncTracker {
     // Executor for expensive tasks
     // We keep the number of max threads very low to have throttling for expensive tasks
     private ExecutorService expensiveExecutor =
-        ExecutorUtil.newMDCAwareCachedThreadPool(
-            5,
-            Integer.MAX_VALUE,
-            new SolrNamedThreadFactory("parallelCoreAdminAPIExpensiveExecutor"));
+        ExecutorUtil.newMDCAwareFixedThreadPool(
+            5, Integer.MAX_VALUE, "parallelCoreAdminAPIExpensiveExecutor");
 
     public CoreAdminAsyncTracker() {
       this(

solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java

@@ -22,7 +22,6 @@
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import org.apache.solr.client.solrj.impl.SolrHttpConstants;
 import org.apache.solr.client.solrj.jetty.HttpJettySolrClient;
@@ -120,17 +119,15 @@ public UpdateShardHandler(UpdateShardHandlerConfig cfg) {
 
     recoveryOnlyClient = recoveryOnlyClientBuilder.build();
 
-    ThreadFactory recoveryThreadFactory = new SolrNamedThreadFactory("recoveryExecutor");
-    if (cfg != null && cfg.getMaxRecoveryThreads() > 0) {
-      if (log.isDebugEnabled()) {
-        log.debug("Creating recoveryExecutor with pool size {}", cfg.getMaxRecoveryThreads());
-      }
+    int maxRecoveryThreads = cfg == null ? -1 : cfg.getMaxRecoveryThreads();
+    if (maxRecoveryThreads > 0) {
+      log.debug("Creating recoveryExecutor with pool size {}", maxRecoveryThreads);
       recoveryExecutor =
           ExecutorUtil.newMDCAwareFixedThreadPool(
-              cfg.getMaxRecoveryThreads(), recoveryThreadFactory);
+              maxRecoveryThreads, new SolrNamedThreadFactory("recoveryExecutor"));
     } else {
       log.debug("Creating recoveryExecutor with unbounded pool");
-      recoveryExecutor = ExecutorUtil.newMDCAwareCachedThreadPool(recoveryThreadFactory);
+      recoveryExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("recoveryExecutor");
     }
   }
 

solr/core/src/test/org/apache/solr/handler/TestHttpRequestId.java

@@ -55,33 +55,33 @@ public static void beforeTest() throws Exception {
   public void mdcContextTest() {
     String collection = "/collection1";
     BlockingQueue<Runnable> workQueue = new SynchronousQueue<>(false);
-    setupClientAndRun(collection, workQueue, 0);
+    setupClientAndRun(collection, workQueue, 0, Integer.MAX_VALUE);
   }
 
   @Test
   public void mdcContextFailureTest() {
     String collection = "/doesnotexist";
     BlockingQueue<Runnable> workQueue = new SynchronousQueue<>(false);
-    setupClientAndRun(collection, workQueue, 0);
+    setupClientAndRun(collection, workQueue, 0, Integer.MAX_VALUE);
   }
 
   @Test
   public void mdcContextTest2() {
     String collection = "/collection1";
     BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(10, false);
-    setupClientAndRun(collection, workQueue, 3);
+    setupClientAndRun(collection, workQueue, 3, 3);
   }
 
   @Test
   public void mdcContextFailureTest2() {
     String collection = "/doesnotexist";
     BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(10, false);
-    setupClientAndRun(collection, workQueue, 3);
+    setupClientAndRun(collection, workQueue, 3, 3);
   }
 
   @SuppressForbidden(reason = "We need to use log4J2 classes directly to test MDC impacts")
   private void setupClientAndRun(
-      String collection, BlockingQueue<Runnable> workQueue, int corePoolSize) {
+      String collection, BlockingQueue<Runnable> workQueue, int corePoolSize, int maximumPoolSize) {
     final String key = "mdcContextTestKey" + System.nanoTime();
     final String value = "TestHttpRequestId" + System.nanoTime();
 
@@ -91,7 +91,7 @@ private void setupClientAndRun(
       ThreadPoolExecutor commExecutor =
           new ExecutorUtil.MDCAwareThreadPoolExecutor(
               corePoolSize,
-              Integer.MAX_VALUE,
+              maximumPoolSize,
               1,
               TimeUnit.SECONDS,
               workQueue,

solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java

@@ -281,22 +281,17 @@ public void testTakeDoesNotHangUnderAsyncInnerFutureCompletion() throws Exceptio
     ExecutorService mockIoThreads =
         ExecutorUtil.newMDCAwareFixedThreadPool(2, new SolrNamedThreadFactory("testMockIo"));
 
-    ExecutorService takeExecutor =
-        ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("testTakeRunner"));
-
+    ExecutorService takeExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("testTakeRunner");
     try {
       for (int i = 0; i < iterations; i++) {
         runAsyncRaceCycle(commExecutor, mockIoThreads, takeExecutor, i, perIterationTimeoutMs);
       }
     } finally {
-      takeExecutor.shutdownNow();
-      takeExecutor.awaitTermination(5, TimeUnit.SECONDS);
-      mockIoThreads.shutdownNow();
-      mockIoThreads.awaitTermination(5, TimeUnit.SECONDS);
+      ExecutorUtil.shutdownNowAndAwaitTermination(takeExecutor);
+      ExecutorUtil.shutdownNowAndAwaitTermination(mockIoThreads);
       commExecutor.shutdown();
       if (!commExecutor.awaitTermination(15, TimeUnit.SECONDS)) {
-        commExecutor.shutdownNow();
-        commExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        ExecutorUtil.shutdownNowAndAwaitTermination(commExecutor);
       }
     }
   }

solr/core/src/test/org/apache/solr/request/TestUnInvertedFieldException.java

@@ -84,10 +84,11 @@ public void testConcurrentInit() throws Exception {
       initCallables.add(() -> UnInvertedField.getUnInvertedField(proto.field(), searcher));
     }
 
+    final int numThreads = TestUtil.nextInt(random(), 3, 6);
     final ThreadPoolExecutor pool =
         new MDCAwareThreadPoolExecutor(
-            3,
-            TestUtil.nextInt(random(), 3, 6),
+            numThreads,
+            numThreads,
             10,
             TimeUnit.MILLISECONDS,
             new LinkedBlockingQueue<Runnable>(),

solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java

@@ -59,7 +59,6 @@
 import org.apache.solr.common.params.SolrParams;
 import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.common.util.TimeSource;
 import org.apache.solr.embedded.JettySolrRunner;
 import org.apache.solr.index.NoMergePolicyFactory;
@@ -343,9 +342,7 @@ private void reorderedDBQsSimpleTest() throws Exception {
     }
 
     // Reordering needs to happen using parallel threads
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
 
     // re-order the updates for NONLEADER 0
     List<UpdateRequest> reorderedUpdates = new ArrayList<>(updates);
@@ -418,9 +415,7 @@ private void reorderedDBQIndividualReplicaTest() throws Exception {
             "inplace_updatable_float:" + (newinplace_updatable_float + 1), version0 + 3));
 
     // Reordering needs to happen using parallel threads
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
 
     // re-order the updates by swapping the last two
     List<UpdateRequest> reorderedUpdates = new ArrayList<>(updates);
@@ -932,9 +927,7 @@ private void outOfOrderUpdatesIndividualReplicaTest() throws Exception {
 
     // Reordering needs to happen using parallel threads, since some of these updates will
     // be blocking calls, waiting for some previous update operations to arrive on which it depends.
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
 
     // re-order the updates for NONLEADER 0
     List<UpdateRequest> reorderedUpdates = new ArrayList<>(updates);
@@ -1036,9 +1029,7 @@ private void reorderedDeletesTest() throws Exception {
     }
 
     // Reordering needs to happen using parallel threads
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
 
     // re-order the updates for NONLEADER 0
     List<UpdateRequest> reorderedUpdates = new ArrayList<>(updates);
@@ -1143,9 +1134,7 @@ private void reorderedDBQsResurrectionTest() throws Exception {
     }
 
     // Reordering needs to happen using parallel threads
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
     // re-order the last two updates for NONLEADER 0
     List<UpdateRequest> reorderedUpdates = new ArrayList<>(updates);
     Collections.swap(reorderedUpdates, 2, 3);
@@ -1248,9 +1237,7 @@ private void delayedReorderingFetchesMissingUpdateFromLeaderTest() throws Except
         .getFilter(ServletFixtures.DelayServlet.class)
         .addDelay("Waiting for dependant update to timeout", 1, 6000);
 
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
     for (UpdateRequest update : updates) {
       AsyncUpdateWithRandomCommit task =
           new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong());
@@ -1353,9 +1340,7 @@ private void delayedReorderingFetchesMissingUpdateFromLeaderTest() throws Except
           .getFilter(ServletFixtures.DelayServlet.class)
           .addDelay("Waiting for dependant update to timeout", 4, 5998); // the delete update
 
-      threadpool =
-          ExecutorUtil.newMDCAwareFixedThreadPool(
-              updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+      threadpool = newThreadPerUpdatePool(updates);
       for (UpdateRequest update : updates) {
         AsyncUpdateWithRandomCommit task =
             new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong());
@@ -1419,6 +1404,10 @@ public void checkExpectedSchemaField(Map<String, Object> expected) throws Except
     assertEquals("Field: " + fieldName, expected, rsp.getField());
   }
 
+  private ExecutorService newThreadPerUpdatePool(List<?> updates) {
+    return ExecutorUtil.newMDCAwareFixedThreadPool(updates.size() + 1, 1, getTestName());
+  }
+
   private class AsyncUpdateWithRandomCommit implements Callable<UpdateResponse> {
     UpdateRequest update;
     SolrClient solrClient;
@@ -1650,9 +1639,7 @@ private void reorderedDBQsUsingUpdatedValueFromADroppedUpdate() throws Exception
         .getFilter(ServletFixtures.DelayServlet.class)
         .addDelay("Waiting for dependant update to timeout", 2, 8000);
 
-    ExecutorService threadpool =
-        ExecutorUtil.newMDCAwareFixedThreadPool(
-            updates.size() + 1, new SolrNamedThreadFactory(getTestName()));
+    ExecutorService threadpool = newThreadPerUpdatePool(updates);
     for (UpdateRequest update : updates) {
       AsyncUpdateWithRandomCommit task =
           new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong());

solr/core/src/test/org/apache/solr/util/OrderedExecutorTest.java

@@ -30,7 +30,6 @@
 import java.util.concurrent.TimeoutException;
 import org.apache.solr.SolrTestCase;
 import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,10 +41,10 @@ private static OrderedExecutor<Object> newOrderedExecutor(int numThreads) {
     // initialize exactly as done in CoreContainer so we test realistically
     return new OrderedExecutor<>(
         numThreads,
-        ExecutorUtil.newMDCAwareCachedThreadPool(
+        ExecutorUtil.newMDCAwareFixedThreadPool(
             numThreads, // thread count
             numThreads, // queue size
-            new SolrNamedThreadFactory("testOrderedExecutor")));
+            "testOrderedExecutor"));
   }
 
   @Test

solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java

@@ -55,7 +55,6 @@
 import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.NamedList;
 import org.apache.solr.common.util.ObjectReleaseTracker;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.eclipse.jetty.client.AuthenticationStore;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.HttpClientTransport;
@@ -83,7 +82,6 @@
 import org.eclipse.jetty.http2.client.HTTP2Client;
 import org.eclipse.jetty.http2.client.transport.HttpClientTransportOverHTTP2;
 import org.eclipse.jetty.io.ClientConnector;
-import org.eclipse.jetty.util.BlockingArrayQueue;
 import org.eclipse.jetty.util.ssl.KeyStoreScanner;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.slf4j.Logger;
@@ -220,10 +218,7 @@ public HttpClient getHttpClient() {
   private HttpClient createHttpClient(Builder builder) {
     executor = builder.getExecutor();
     if (executor == null) {
-      BlockingArrayQueue<Runnable> queue = BlockingArrayQueue.newInstance(256, Integer.MAX_VALUE);
-      this.executor =
-          new ExecutorUtil.MDCAwareThreadPoolExecutor(
-              32, 256, 60, TimeUnit.SECONDS, queue, new SolrNamedThreadFactory("h2sc"));
+      this.executor = ExecutorUtil.newMDCAwareCachedThreadPool("h2sc"); // nocommmit new PR
       shutdownExecutor = true;
     } else {
       shutdownExecutor = false;