Rationalize thread pool executor usage

4 thread pools:
 - executor service for internal tasks (e.g. run connection recovery)
 - scheduled executor service for internal scheduled tasks (e.g.
   schedule a connection retry)
 - publisher executor service to poll settlement futures
 - consumer executor service to poll receivers

 The tasks to run are different in nature, it explains the different
 executor services to configure. The library uses sensible defaults (e.g
 virtual threads if available for the publisher executor service, where
 there is a short task for each message; platform threads for the
 consumer executor, where tasks are long-running).

 Virtual threads do not seem appropriate for long-running tasks,
 especially when testing against cluster rolling restart.

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -558,10 +558,6 @@ AmqpEnvironment environment() {
     return this.environment;
   }
 
-  ExecutorService executorService() {
-    return this.environment.executorService();
-  }
-
   ScheduledExecutorService scheduledExecutorService() {
     return this.environment.scheduledExecutorService();
   }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -206,7 +206,7 @@ private Runnable createReceiveTask(Receiver receiver, MessageHandler messageHand
 
   private void startReceivingLoop() {
     Runnable receiveTask = createReceiveTask(nativeReceiver, messageHandler);
-    this.receiveLoop = this.connection.executorService().submit(receiveTask);
+    this.receiveLoop = this.connection.environment().consumerExecutorService().submit(receiveTask);
   }
 
   void recoverAfterConnectionFailure() {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironment.java

@@ -32,28 +32,33 @@ class AmqpEnvironment implements Environment {
   private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);
 
   private final Client client;
-  private final ExecutorService executorService;
   private final DefaultConnectionSettings<?> connectionSettings =
       DefaultConnectionSettings.instance();
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final boolean internalExecutor;
   private final boolean internalScheduledExecutor;
+  private final boolean internalConsumerExecutor;
+  private final boolean internalPublisherExecutor;
+  private final ExecutorService executorService;
+  private final ScheduledExecutorService scheduledExecutorService;
+  private final ExecutorService publisherExecutorService;
+  private final ExecutorService consumerExecutorService;
   private final ConnectionManager connectionManager = new ConnectionManager(this);
   private final long id;
   private final Clock clock = new Clock();
-  private final ScheduledExecutorService scheduledExecutorService;
   private volatile ScheduledFuture<?> clockRefreshFuture;
   private final AtomicBoolean clockRefreshSet = new AtomicBoolean(false);
   private final MetricsCollector metricsCollector;
   private final ObservationCollector observationCollector;
   private final ConnectionUtils.AffinityCache affinityCache = new ConnectionUtils.AffinityCache();
   private final EventLoop recoveryEventLoop;
   private final ExecutorService recoveryEventLoopExecutorService;
-  private final ExecutorService managementExecutorService;
 
   AmqpEnvironment(
       ExecutorService executorService,
       ScheduledExecutorService scheduledExecutorService,
+      ExecutorService publisherExecutorService,
+      ExecutorService consumerExecutorService,
       DefaultConnectionSettings<?> connectionSettings,
       MetricsCollector metricsCollector,
       ObservationCollector observationCollector) {
@@ -65,7 +70,7 @@ class AmqpEnvironment implements Environment {
 
     String threadPrefix = String.format("rabbitmq-amqp-environment-%d-", this.id);
     if (executorService == null) {
-      this.executorService = Utils.executorService(threadPrefix);
+      this.executorService = Executors.newCachedThreadPool(Utils.threadFactory(threadPrefix));
       this.internalExecutor = true;
     } else {
       this.executorService = executorService;
@@ -79,6 +84,21 @@ class AmqpEnvironment implements Environment {
       this.scheduledExecutorService = scheduledExecutorService;
       this.internalScheduledExecutor = false;
     }
+    if (publisherExecutorService == null) {
+      this.publisherExecutorService = Utils.executorService(threadPrefix);
+      this.internalPublisherExecutor = true;
+    } else {
+      this.publisherExecutorService = publisherExecutorService;
+      this.internalPublisherExecutor = false;
+    }
+    if (consumerExecutorService == null) {
+      this.consumerExecutorService =
+          Executors.newCachedThreadPool(Utils.threadFactory(threadPrefix + "consumer-"));
+      this.internalConsumerExecutor = true;
+    } else {
+      this.consumerExecutorService = consumerExecutorService;
+      this.internalConsumerExecutor = false;
+    }
     this.metricsCollector =
         metricsCollector == null ? NoOpMetricsCollector.INSTANCE : metricsCollector;
     this.observationCollector =
@@ -92,14 +112,6 @@ class AmqpEnvironment implements Environment {
             new LinkedBlockingQueue<>(),
             Utils.threadFactory(threadPrefix + "event-loop-"));
     this.recoveryEventLoop = new EventLoop(this.recoveryEventLoopExecutorService);
-    this.managementExecutorService =
-        new ThreadPoolExecutor(
-            0,
-            Integer.MAX_VALUE,
-            30,
-            TimeUnit.SECONDS,
-            new SynchronousQueue<>(),
-            Utils.threadFactory(threadPrefix + "management-loop-"));
   }
 
   DefaultConnectionSettings<?> connectionSettings() {
@@ -111,6 +123,7 @@ Client client() {
   }
 
   Clock clock() {
+    // FIXME set clock on demand
     if (this.clockRefreshSet.compareAndSet(false, true)) {
       this.clockRefreshFuture =
           this.scheduledExecutorService.scheduleAtFixedRate(
@@ -126,13 +139,18 @@ public void close() {
       this.client.close();
       this.recoveryEventLoop.close();
       this.recoveryEventLoopExecutorService.shutdownNow();
-      this.managementExecutorService.shutdownNow();
       if (this.internalExecutor) {
         this.executorService.shutdownNow();
       }
       if (this.internalScheduledExecutor) {
         this.scheduledExecutorService.shutdownNow();
       }
+      if (this.internalPublisherExecutor) {
+        this.publisherExecutorService.shutdownNow();
+      }
+      if (this.internalConsumerExecutor) {
+        this.consumerExecutorService.shutdownNow();
+      }
       if (this.clockRefreshFuture != null) {
         this.clockRefreshFuture.cancel(false);
       }
@@ -149,8 +167,12 @@ ExecutorService executorService() {
     return this.executorService;
   }
 
-  ExecutorService managementExecutorService() {
-    return this.managementExecutorService;
+  ExecutorService publisherExecutorService() {
+    return this.publisherExecutorService;
+  }
+
+  ExecutorService consumerExecutorService() {
+    return this.consumerExecutorService;
   }
 
   ScheduledExecutorService scheduledExecutorService() {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironmentBuilder.java

@@ -33,6 +33,8 @@ public class AmqpEnvironmentBuilder implements EnvironmentBuilder {
       new DefaultEnvironmentConnectionSettings(this);
   private ExecutorService executorService;
   private ScheduledExecutorService scheduledExecutorService;
+  private ExecutorService publisherExecutorService;
+  private ExecutorService consumerExecutorService;
   private MetricsCollector metricsCollector = NoOpMetricsCollector.INSTANCE;
   private ObservationCollector observationCollector = Utils.NO_OP_OBSERVATION_COLLECTOR;
 
@@ -49,6 +51,16 @@ public AmqpEnvironmentBuilder scheduledExecutorService(
     return this;
   }
 
+  public AmqpEnvironmentBuilder publisherExecutorService(ExecutorService publisherExecutorService) {
+    this.publisherExecutorService = publisherExecutorService;
+    return this;
+  }
+
+  public AmqpEnvironmentBuilder consumerExecutorService(ExecutorService consumerExecutorService) {
+    this.consumerExecutorService = consumerExecutorService;
+    return this;
+  }
+
   public AmqpEnvironmentBuilder metricsCollector(MetricsCollector metricsCollector) {
     this.metricsCollector = metricsCollector;
     return this;
@@ -69,6 +81,8 @@ public Environment build() {
     return new AmqpEnvironment(
         executorService,
         scheduledExecutorService,
+        publisherExecutorService,
+        consumerExecutorService,
         connectionSettings,
         metricsCollector,
         observationCollector);

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -368,8 +368,7 @@ OutstandingRequest request(Message<?> request, UUID requestId) throws ClientExce
         if (loop == null) {
           Runnable receiveTask = receiveTask();
           LOGGER.debug("Starting management receive loop ({}).", this);
-          this.receiveLoop =
-              this.connection.environment().managementExecutorService().submit(receiveTask);
+          this.receiveLoop = this.connection.environment().executorService().submit(receiveTask);
           LOGGER.debug("Management initialized ({}).", this);
         }
       } finally {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpPublisher.java

@@ -57,7 +57,7 @@ final class AmqpPublisher extends ResourceBase implements Publisher {
   AmqpPublisher(AmqpPublisherBuilder builder) {
     super(builder.listeners());
     this.id = ID_SEQUENCE.getAndIncrement();
-    this.executorService = builder.connection().executorService();
+    this.executorService = builder.connection().environment().publisherExecutorService();
     this.address = builder.address();
     this.destinationSpec = builder.destination();
     this.connection = builder.connection();

src/test/java/com/rabbitmq/client/amqp/impl/AmqpConnectionRecoveryTest.java

@@ -56,6 +56,8 @@ static void initAll() {
         });
     environment =
         new AmqpEnvironment(
+            null,
+            null,
             null,
             null,
             connectionSettings,

src/test/java/com/rabbitmq/client/amqp/impl/RecoveryClusterTest.java

@@ -34,7 +34,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -69,14 +68,8 @@ static void initAll() {
 
   @BeforeEach
   void init(TestInfo info) {
-    executorService = Executors.newCachedThreadPool(Utils.threadFactory("recover-cluster-test-"));
     environment =
-        new AmqpEnvironmentBuilder()
-            .connectionSettings()
-            .uris(URIS)
-            .environmentBuilder()
-            .executorService(executorService)
-            .build();
+        new AmqpEnvironmentBuilder().connectionSettings().uris(URIS).environmentBuilder().build();
     this.connection = connection(b -> b.name("c-management").recovery().connectionBuilder());
     this.management = connection.management();
     this.testInfo = info;