Remove consumer executor service

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

@@ -37,12 +37,10 @@ class AmqpEnvironment implements Environment {
   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();
@@ -58,7 +56,6 @@ class AmqpEnvironment implements Environment {
       ExecutorService executorService,
       ScheduledExecutorService scheduledExecutorService,
       ExecutorService publisherExecutorService,
-      ExecutorService consumerExecutorService,
       DefaultConnectionSettings<?> connectionSettings,
       MetricsCollector metricsCollector,
       ObservationCollector observationCollector) {
@@ -91,14 +88,6 @@ class AmqpEnvironment implements Environment {
       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 =
@@ -147,9 +136,6 @@ public void close() {
       if (this.internalPublisherExecutor) {
         this.publisherExecutorService.shutdownNow();
       }
-      if (this.internalConsumerExecutor) {
-        this.consumerExecutorService.shutdownNow();
-      }
       if (this.clockRefreshFuture != null) {
         this.clockRefreshFuture.cancel(false);
       }
@@ -170,10 +156,6 @@ ExecutorService publisherExecutorService() {
     return this.publisherExecutorService;
   }
 
-  ExecutorService consumerExecutorService() {
-    return this.consumerExecutorService;
-  }
-
   ScheduledExecutorService scheduledExecutorService() {
     return this.scheduledExecutorService;
   }

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

@@ -35,7 +35,6 @@ public class AmqpEnvironmentBuilder implements EnvironmentBuilder {
   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;
 
@@ -82,15 +81,18 @@ public AmqpEnvironmentBuilder publisherExecutorService(ExecutorService publisher
   }
 
   /**
-   * Set executor service used for consumer loops.
+   * Deprecated, do not use anymore. Consumers do not use a polling loop anymore.
+   *
+   * <p>Set executor service used for consumer loops.
    *
    * <p>The library uses sensible defaults, override only in case of problems.
    *
    * @param consumerExecutorService the executor service
    * @return this builder instance
+   * @deprecated Do not use anymore
    */
+  @Deprecated(forRemoval = true)
   public AmqpEnvironmentBuilder consumerExecutorService(ExecutorService consumerExecutorService) {
-    this.consumerExecutorService = consumerExecutorService;
     return this;
   }
 
@@ -144,7 +146,6 @@ public Environment build() {
         executorService,
         scheduledExecutorService,
         publisherExecutorService,
-        consumerExecutorService,
         connectionSettings,
         metricsCollector,
         observationCollector);

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

@@ -59,7 +59,6 @@ static void initAll() {
             null,
             null,
             null,
-            null,
             connectionSettings,
             NoOpMetricsCollector.INSTANCE,
             Utils.NO_OP_OBSERVATION_COLLECTOR);