Remove consumer executor service

It was used for consumer polling loops, but consumers do not work this
way anymore.

The method to set it in the environment builder has been deprecated and
is now a no-op.

References #93

Author: acogoluegnes

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);