Dispatch connection shutdown in appropriate threads

For Netty. Make sure that connection shutdown sequence is not executed
in the IO event loop if recovery follows. Recovery kicks in the shutdown
sequence, so we could end up with a deadlock if the new connection is
allocated to the same event loop.

References #1663

Author: acogoluegnes

pom.xml

@@ -776,6 +776,7 @@
             <includes>
               <include>src/main/java/com/rabbitmq/client/ConnectionFactory.java</include>
               <include>src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java</include>
+              <include>src/main/java/com/rabbitmq/client/impl/Environment.java</include>
               <include>src/main/java/com/rabbitmq/client/observation/**/*.java</include>
               <include>src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java</include>
               <include>src/test/java/com/rabbitmq/client/test/NettyTest.java</include>

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -1063,6 +1063,10 @@ public ConnectionFactory setCredentialsRefreshService(
   protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IOException {
     if (netty) {
       if (this.frameHandlerFactory == null) {
+        Predicate<ShutdownSignalException> recoveryCondition =
+            this.connectionRecoveryTriggeringCondition == null
+                ? AutorecoveringConnection.DEFAULT_CONNECTION_RECOVERY_TRIGGERING_CONDITION
+                : this.connectionRecoveryTriggeringCondition;
         this.frameHandlerFactory =
             new NettyFrameHandlerFactory(
                 this.nettyConf.eventLoopGroup,
@@ -1072,7 +1076,9 @@ protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IO
                 this.nettyConf.enqueuingTimeout,
                 connectionTimeout,
                 socketConf,
-                maxInboundMessageBodySize);
+                maxInboundMessageBodySize,
+                this.automaticRecovery,
+                recoveryCondition);
       }
       return this.frameHandlerFactory;
     } else {

src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java

@@ -442,6 +442,7 @@ private ChannelState(Channel channel) {
      *
      * @deprecated Use {@link #markRejectedMessage(boolean)} instead
      */
+    @Deprecated
     protected abstract void markRejectedMessage();
 
     /**

src/main/java/com/rabbitmq/client/impl/Environment.java

@@ -1,4 +1,5 @@
-// Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2025 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -12,40 +13,45 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // info@rabbitmq.com.
-
 package com.rabbitmq.client.impl;
 
+import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 
 /**
- * Infers information about the execution environment, e.g.
- * security permissions.
- * Package-protected API.
+ * Infers information about the execution environment, e.g. security permissions. Package-protected
+ * API.
  */
 public class Environment {
 
-    /**
-     * This method is deprecated and subject to removal in the next major release.
-     *
-     * There is no replacement for this method, as it used to use the
-     * {@link SecurityManager}, which is itself deprecated and subject to removal.
-     * @deprecated
-     * @return always returns true
-     */
-    @Deprecated
-    public static boolean isAllowedToModifyThreads() {
-       return true;
-    }
+  /**
+   * This method is deprecated and subject to removal in the next major release.
+   *
+   * <p>There is no replacement for this method, as it used to use the {@link SecurityManager},
+   * which is itself deprecated and subject to removal.
+   *
+   * @deprecated
+   * @return always returns true
+   */
+  @Deprecated
+  public static boolean isAllowedToModifyThreads() {
+    return true;
+  }
+
+  static Thread newThread(Runnable runnable, String name) {
+    return newThread(Executors.defaultThreadFactory(), runnable, name);
+  }
 
-    public static Thread newThread(ThreadFactory factory, Runnable runnable, String name) {
-        Thread t = factory.newThread(runnable);
-        t.setName(name);
-        return t;
-    }
+  public static Thread newThread(ThreadFactory factory, Runnable runnable, String name) {
+    Thread t = factory.newThread(runnable);
+    t.setName(name);
+    return t;
+  }
 
-    public static Thread newThread(ThreadFactory factory, Runnable runnable, String name, boolean isDaemon) {
-        Thread t = newThread(factory, runnable, name);
-        t.setDaemon(isDaemon);
-        return t;
-    }
+  public static Thread newThread(
+      ThreadFactory factory, Runnable runnable, String name, boolean isDaemon) {
+    Thread t = newThread(factory, runnable, name);
+    t.setDaemon(isDaemon);
+    return t;
+  }
 }

src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

@@ -23,6 +23,7 @@
 import com.rabbitmq.client.Address;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MalformedFrameException;
+import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.SocketConfigurator;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
@@ -61,6 +62,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import javax.net.ssl.SSLHandshakeException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,6 +75,7 @@ public final class NettyFrameHandlerFactory extends AbstractFrameHandlerFactory
   private final Consumer<Channel> channelCustomizer;
   private final Consumer<Bootstrap> bootstrapCustomizer;
   private final Duration enqueuingTimeout;
+  private final Predicate<ShutdownSignalException> willRecover;
 
   public NettyFrameHandlerFactory(
       EventLoopGroup eventLoopGroup,
@@ -82,14 +85,30 @@ public NettyFrameHandlerFactory(
       Duration enqueuingTimeout,
       int connectionTimeout,
       SocketConfigurator configurator,
-      int maxInboundMessageBodySize) {
+      int maxInboundMessageBodySize,
+      boolean automaticRecovery,
+      Predicate<ShutdownSignalException> recoveryCondition) {
     super(connectionTimeout, configurator, sslContextFactory != null, maxInboundMessageBodySize);
     this.eventLoopGroup = eventLoopGroup;
     this.sslContextFactory = sslContextFactory == null ? connName -> null : sslContextFactory;
     this.channelCustomizer = channelCustomizer == null ? Utils.noOpConsumer() : channelCustomizer;
     this.bootstrapCustomizer =
         bootstrapCustomizer == null ? Utils.noOpConsumer() : bootstrapCustomizer;
     this.enqueuingTimeout = enqueuingTimeout;
+    this.willRecover =
+        sse -> {
+          if (!automaticRecovery) {
+            return false;
+          } else {
+            try {
+              return recoveryCondition.test(sse);
+            } catch (Exception e) {
+              // we assume it will recover, so we take the safe path to dispatch the closing
+              // it avoids the risk of deadlock
+              return true;
+            }
+          }
+        };
   }
 
   private static void closeNettyState(Channel channel, EventLoopGroup eventLoopGroup) {
@@ -133,6 +152,7 @@ public FrameHandler create(Address addr, String connectionName) throws IOExcepti
         sslContext,
         this.eventLoopGroup,
         this.enqueuingTimeout,
+        this.willRecover,
         this.channelCustomizer,
         this.bootstrapCustomizer);
   }
@@ -163,6 +183,7 @@ private NettyFrameHandler(
         SslContext sslContext,
         EventLoopGroup elg,
         Duration enqueuingTimeout,
+        Predicate<ShutdownSignalException> willRecover,
         Consumer<Channel> channelCustomizer,
         Consumer<Bootstrap> bootstrapCustomizer)
         throws IOException {
@@ -195,7 +216,8 @@ private NettyFrameHandler(
       int lengthFieldOffset = 3;
       int lengthFieldLength = 4;
       int lengthAdjustement = 1;
-      AmqpHandler amqpHandler = new AmqpHandler(maxInboundMessageBodySize, this::close);
+      AmqpHandler amqpHandler =
+          new AmqpHandler(maxInboundMessageBodySize, this::close, willRecover);
       int port = ConnectionFactory.portOrDefault(addr.getPort(), sslContext != null);
       b.handler(
           new ChannelInitializer<SocketChannel>() {
@@ -404,14 +426,26 @@ private static class AmqpHandler extends ChannelInboundHandlerAdapter {
 
     private final int maxPayloadSize;
     private final Runnable closeSequence;
+    private final Predicate<ShutdownSignalException> willRecover;
     private volatile AMQConnection connection;
+    private volatile Channel ch;
     private final AtomicBoolean writable = new AtomicBoolean(true);
     private final AtomicReference<CountDownLatch> writableLatch =
         new AtomicReference<>(new CountDownLatch(1));
 
-    private AmqpHandler(int maxPayloadSize, Runnable closeSequence) {
+    private AmqpHandler(
+        int maxPayloadSize,
+        Runnable closeSequence,
+        Predicate<ShutdownSignalException> willRecover) {
       this.maxPayloadSize = maxPayloadSize;
       this.closeSequence = closeSequence;
+      this.willRecover = willRecover;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+      this.ch = ctx.channel();
+      super.channelActive(ctx);
     }
 
     @Override
@@ -444,7 +478,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         if (noProblem
             && (!this.connection.isRunning() || this.connection.hasBrokerInitiatedShutdown())) {
           // looks like the frame was Close-Ok or Close
-          ctx.executor().submit(() -> this.connection.doFinalShutdown());
+          this.dispatchShutdownToConnection(() -> this.connection.doFinalShutdown());
         }
       } finally {
         m.release();
@@ -504,10 +538,10 @@ public void channelInactive(ChannelHandlerContext ctx) {
         AMQConnection c = this.connection;
         if (c.isOpen()) {
           // it is likely to be an IO exception
-          c.handleIoError(null);
+          this.dispatchShutdownToConnection(() -> c.handleIoError(null));
         } else {
           // just in case, the call is idempotent anyway
-          c.doFinalShutdown();
+          this.dispatchShutdownToConnection(c::doFinalShutdown);
         }
       }
     }
@@ -533,7 +567,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
               this.connection.getAddress().getHostName(),
               this.connection.getPort());
           if (needToDispatchIoError()) {
-            this.connection.handleHeartbeatFailure();
+            this.dispatchShutdownToConnection(() -> this.connection.handleHeartbeatFailure());
           }
         } else if (e.state() == IdleState.WRITER_IDLE) {
           this.connection.writeFrame(new Frame(AMQP.FRAME_HEARTBEAT, 0));
@@ -545,7 +579,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
 
     private void handleIoError(Throwable cause) {
       if (needToDispatchIoError()) {
-        this.connection.handleIoError(cause);
+        this.dispatchShutdownToConnection(() -> this.connection.handleIoError(cause));
       } else {
         this.closeSequence.run();
       }
@@ -563,6 +597,32 @@ private boolean isWritable() {
     private CountDownLatch writableLatch() {
       return this.writableLatch.get();
     }
+
+    protected void dispatchShutdownToConnection(Runnable connectionShutdownRunnable) {
+      String name = "rabbitmq-connection-shutdown";
+      AMQConnection c = this.connection;
+      if (c == null || ch == null) {
+        // not enough information, we dispatch in separate thread
+        Environment.newThread(connectionShutdownRunnable, name).start();
+      } else {
+        if (ch.eventLoop().inEventLoop()) {
+          if (this.willRecover.test(c.getCloseReason())) {
+            // the connection will recover, we don't want this to happen in the event loop,
+            // it could cause a deadlock, so using a separate thread
+            name = name + "-" + c;
+            System.out.println("in separate thread");
+            Environment.newThread(connectionShutdownRunnable, name).start();
+          } else {
+            // no recovery, it is safe to dispatch in the event loop
+            System.out.println("in event loop");
+            ch.eventLoop().submit(connectionShutdownRunnable);
+          }
+        } else {
+          // not in the event loop, we can run it in the same thread
+          connectionShutdownRunnable.run();
+        }
+      }
+    }
   }
 
   private static final class ProtocolVersionMismatchHandler extends ChannelInboundHandlerAdapter {

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

@@ -144,7 +144,6 @@ public void afterAll(ExtensionContext context) {
             try {
               eventLoopGroup.shutdownGracefully(0, 0, SECONDS).get(10, SECONDS);
             } catch (InterruptedException e) {
-              LOGGER.debug("Error while asynchronously closing Netty event loop group", e);
               Thread.currentThread().interrupt();
             } catch (Exception e) {
               LOGGER.warn("Error while asynchronously closing Netty event loop group", e);