Add dedicated executor to close connection in NIO mode

src/main/java/com/rabbitmq/client/impl/nio/NioLoop.java

@@ -41,9 +41,12 @@ public class NioLoop implements Runnable {
 
     private final NioParams nioParams;
 
+    private final ExecutorService connectionShutdownExecutor;
+
     public NioLoop(NioParams nioParams, NioLoopContext loopContext) {
         this.nioParams = nioParams;
         this.context = loopContext;
+        this.connectionShutdownExecutor = nioParams.getConnectionShutdownExecutor();
     }
 
     @Override
@@ -283,19 +286,15 @@ protected void dispatchIoErrorToConnection(final SocketChannelFrameHandlerState
     }
 
     protected void dispatchShutdownToConnection(final SocketChannelFrameHandlerState state) {
-        Runnable shutdown = new Runnable() {
-
-            @Override
-            public void run() {
-                state.getConnection().doFinalShutdown();
-            }
-        };
-        if (executorService() == null) {
+        Runnable shutdown = () -> state.getConnection().doFinalShutdown();
+        if (this.connectionShutdownExecutor != null) {
+            connectionShutdownExecutor.execute(shutdown);
+        } else if (executorService() != null) {
+            executorService().execute(shutdown);
+        } else {
             String name = "rabbitmq-connection-shutdown-" + state.getConnection();
             Thread shutdownThread = Environment.newThread(threadFactory(), shutdown, name);
             shutdownThread.start();
-        } else {
-            executorService().submit(shutdown);
         }
     }
 

src/main/java/com/rabbitmq/client/impl/nio/NioParams.java

@@ -55,10 +55,10 @@ public class NioParams {
     private SocketChannelConfigurator socketChannelConfigurator = new DefaultSocketChannelConfigurator();
 
     /** the hook to configure the SSL engine before the connection is open */
-    private SslEngineConfigurator sslEngineConfigurator = new SslEngineConfigurator() {
-        @Override
-        public void configure(SSLEngine sslEngine) throws IOException { }
-    };
+    private SslEngineConfigurator sslEngineConfigurator = sslEngine -> { };
+
+    /** the executor service used for connection shutdown */
+    private ExecutorService connectionShutdownExecutor;
 
     public NioParams() {
     }
@@ -72,6 +72,7 @@ public NioParams(NioParams nioParams) {
         setNioExecutor(nioParams.getNioExecutor());
         setThreadFactory(nioParams.getThreadFactory());
         setSslEngineConfigurator(nioParams.getSslEngineConfigurator());
+        setConnectionShutdownExecutor(nioParams.getConnectionShutdownExecutor());
     }
 
     public int getReadByteBufferSize() {
@@ -186,6 +187,9 @@ public ExecutorService getNioExecutor() {
      * number of requested IO threads, plus a few more, as it's also
      * used to dispatch the shutdown of connections.
      *
+     * Connection shutdown can also be handled by a dedicated {@link ExecutorService},
+     * see {@link #setConnectionShutdownExecutor(ExecutorService)}.
+     *
      * It's developer's responsibility to shut down the executor
      * when it is no longer needed.
      *
@@ -195,6 +199,7 @@ public ExecutorService getNioExecutor() {
      * @return this {@link NioParams} instance
      * @see NioParams#setNbIoThreads(int)
      * @see NioParams#setThreadFactory(ThreadFactory)
+     * @see NioParams#setConnectionShutdownExecutor(ExecutorService)
      */
     public NioParams setNioExecutor(ExecutorService nioExecutor) {
         this.nioExecutor = nioExecutor;
@@ -275,4 +280,36 @@ public void setSslEngineConfigurator(SslEngineConfigurator configurator) {
     public SslEngineConfigurator getSslEngineConfigurator() {
         return sslEngineConfigurator;
     }
+
+    /**
+     * Set the {@link ExecutorService} used for connection shutdown.
+     * If not set, falls back to the NIO executor and then the thread factory.
+     * This executor service is useful when strict control of the number of threads
+     * is necessary, the application can experience the closing of several connections
+     * at once, and automatic recovery is enabled. In such cases, the connection recovery
+     * can take place in the same pool of threads as the NIO operations, which can
+     * create deadlocks (all the threads of the pool are busy recovering, and there's no
+     * thread left for NIO, so connections never recover).
+     * <p>
+     * Note it's developer's responsibility to shut down the executor
+     * when it is no longer needed.
+     * <p>
+     * Using the thread factory for such scenarios avoid the deadlocks, at the price
+     * of potentially creating many short-lived threads in case of massive connection lost.
+     * <p>
+     * With both the NIO and connection shutdown executor services set and configured
+     * accordingly, the application can control reliably the number of threads used.
+     *
+     * @param connectionShutdownExecutor the executor service to use
+     * @return this {@link NioParams} instance
+     * @see NioParams#setNioExecutor(ExecutorService)
+     */
+    public NioParams setConnectionShutdownExecutor(ExecutorService connectionShutdownExecutor) {
+        this.connectionShutdownExecutor = connectionShutdownExecutor;
+        return this;
+    }
+
+    public ExecutorService getConnectionShutdownExecutor() {
+        return connectionShutdownExecutor;
+    }
 }

src/test/java/com/rabbitmq/client/test/ClientTests.java

@@ -63,7 +63,8 @@
     NoAutoRecoveryWhenTcpWindowIsFullTest.class,
     JsonRpcTest.class,
     AddressTest.class,
-    DefaultRetryHandlerTest.class
+    DefaultRetryHandlerTest.class,
+    NioDeadlockOnConnectionClosing.class
 })
 public class ClientTests {
 

src/test/java/com/rabbitmq/client/test/NioDeadlockOnConnectionClosing.java

@@ -0,0 +1,98 @@
+// Copyright (c) 2007-Present Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.impl.nio.NioParams;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static com.rabbitmq.client.test.TestUtils.closeAllConnectionsAndWaitForRecovery;
+import static org.junit.Assert.assertTrue;
+
+/**
+ *
+ */
+public class NioDeadlockOnConnectionClosing {
+
+    static final Logger LOGGER = LoggerFactory.getLogger(NioDeadlockOnConnectionClosing.class);
+
+    ExecutorService nioExecutorService, connectionShutdownExecutorService;
+    ConnectionFactory cf;
+    List<Connection> connections;
+
+    @Before
+    public void setUp() {
+        nioExecutorService = Executors.newFixedThreadPool(2);
+        connectionShutdownExecutorService = Executors.newFixedThreadPool(2);
+        cf = TestUtils.connectionFactory();
+        cf.setAutomaticRecoveryEnabled(true);
+        cf.useNio();
+        cf.setNetworkRecoveryInterval(1000);
+        NioParams params = new NioParams()
+            .setNioExecutor(nioExecutorService)
+            .setConnectionShutdownExecutor(connectionShutdownExecutorService)
+            .setNbIoThreads(2);
+        cf.setNioParams(params);
+        connections = new ArrayList<>();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        for (Connection connection : connections) {
+            try {
+                connection.close(2000);
+            } catch (Exception e) {
+                LOGGER.warn("Error while closing test connection", e);
+            }
+        }
+
+        shutdownExecutorService(nioExecutorService);
+        shutdownExecutorService(connectionShutdownExecutorService);
+    }
+
+    private void shutdownExecutorService(ExecutorService executorService) throws InterruptedException {
+        if (executorService == null) {
+            return;
+        }
+        executorService.shutdown();
+        boolean terminated = executorService.awaitTermination(5, TimeUnit.SECONDS);
+        if (!terminated) {
+            LOGGER.warn("Couldn't terminate executor after 5 seconds");
+        }
+    }
+
+    @Test
+    public void connectionClosing() throws Exception {
+        for (int i = 0; i < 10; i++) {
+            connections.add(cf.newConnection());
+        }
+        closeAllConnectionsAndWaitForRecovery(connections);
+        for (Connection connection : connections) {
+            assertTrue(connection.isOpen());
+        }
+    }
+}

src/test/java/com/rabbitmq/client/test/TestUtils.java

@@ -31,6 +31,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -156,26 +158,36 @@ public static void closeAndWaitForRecovery(RecoverableConnection connection) thr
         wait(latch);
     }
 
-    public static void closeAllConnectionsAndWaitForRecovery(Connection connection) throws IOException, InterruptedException {
-        CountDownLatch latch = prepareForRecovery(connection);
+    public static void closeAllConnectionsAndWaitForRecovery(Collection<Connection> connections) throws IOException, InterruptedException {
+        CountDownLatch latch = prepareForRecovery(connections);
         Host.closeAllConnections();
         wait(latch);
     }
 
-    public static CountDownLatch prepareForRecovery(Connection conn) {
-        final CountDownLatch latch = new CountDownLatch(1);
-        ((AutorecoveringConnection) conn).addRecoveryListener(new RecoveryListener() {
+    public static void closeAllConnectionsAndWaitForRecovery(Connection connection) throws IOException, InterruptedException {
+        closeAllConnectionsAndWaitForRecovery(Collections.singletonList(connection));
+    }
 
-            @Override
-            public void handleRecovery(Recoverable recoverable) {
-                latch.countDown();
-            }
+    public static CountDownLatch prepareForRecovery(Connection connection) {
+        return prepareForRecovery(Collections.singletonList(connection));
+    }
 
-            @Override
-            public void handleRecoveryStarted(Recoverable recoverable) {
-                // No-op
-            }
-        });
+    public static CountDownLatch prepareForRecovery(Collection<Connection> connections) {
+        final CountDownLatch latch = new CountDownLatch(connections.size());
+        for (Connection conn : connections) {
+            ((AutorecoveringConnection) conn).addRecoveryListener(new RecoveryListener() {
+
+                @Override
+                public void handleRecovery(Recoverable recoverable) {
+                    latch.countDown();
+                }
+
+                @Override
+                public void handleRecoveryStarted(Recoverable recoverable) {
+                    // No-op
+                }
+            });
+        }
         return latch;
     }