Use test class name for thread prefix

(cherry picked from commit 6d98b74)

Author: acogoluegnes

pom.xml

@@ -774,6 +774,7 @@
         <configuration>
           <java>
             <includes>
+              <include>src/main/java/com/rabbitmq/AmqpClientTestExtension.java</include>
               <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>

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

@@ -57,6 +57,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -430,6 +431,8 @@ private static class AmqpHandler extends ChannelInboundHandlerAdapter {
     private final AtomicReference<CountDownLatch> writableLatch =
         new AtomicReference<>(new CountDownLatch(1));
     private final AtomicBoolean shutdownDispatched = new AtomicBoolean(false);
+    private static final AtomicInteger SEQUENCE = new AtomicInteger(0);
+    private final String id;
 
     private AmqpHandler(
         int maxPayloadSize,
@@ -438,6 +441,7 @@ private AmqpHandler(
       this.maxPayloadSize = maxPayloadSize;
       this.closeSequence = closeSequence;
       this.willRecover = willRecover;
+      this.id = "amqp-handler-" + SEQUENCE.getAndIncrement();
     }
 
     @Override
@@ -501,6 +505,7 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio
     public void channelInactive(ChannelHandlerContext ctx) {
       if (needToDispatchIoError()) {
         AMQConnection c = this.connection;
+        LOGGER.debug("Dispatching shutdown when channel became inactive ({})", this.id);
         if (c.isOpen()) {
           // it is likely to be an IO exception
           this.dispatchShutdownToConnection(() -> c.handleIoError(null));
@@ -565,7 +570,7 @@ private CountDownLatch writableLatch() {
 
     protected void dispatchShutdownToConnection(Runnable connectionShutdownRunnable) {
       if (this.shutdownDispatched.compareAndSet(false, true)) {
-        String name = "rabbitmq-connection-shutdown";
+        String name = "rabbitmq-connection-shutdown-" + this.id;
         AMQConnection c = this.connection;
         if (c == null || ch == null) {
           // not enough information, we dispatch in separate thread
@@ -575,7 +580,7 @@ protected void dispatchShutdownToConnection(Runnable connectionShutdownRunnable)
             if (this.willRecover.test(c.getCloseReason()) || ch.eventLoop().isShuttingDown()) {
               // 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;
+              //              name = name + "-" + c;
               Environment.newThread(connectionShutdownRunnable, name).start();
             } else {
               // no recovery, it is safe to dispatch in the event loop

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

@@ -31,6 +31,9 @@
 import java.net.Socket;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
@@ -105,7 +108,8 @@ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext con
   @Override
   public void beforeAll(ExtensionContext context) {
     if (TestUtils.isNetty()) {
-      EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
+      ThreadFactory tf = new NamedThreadFactory(context.getTestClass().get().getSimpleName() + "-");
+      EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(tf, NioIoHandler.newFactory());
       store(context)
           .put("nettyEventLoopGroup", eventLoopGroup);
       TestUtils.eventLoopGroup(eventLoopGroup);
@@ -165,4 +169,29 @@ public void close() {
       this.executorService.shutdownNow();
     }
   }
+
+  private static class NamedThreadFactory implements ThreadFactory {
+
+    private final ThreadFactory backingThreadFactory;
+
+    private final String prefix;
+
+    private final AtomicLong count = new AtomicLong(0);
+
+    private NamedThreadFactory(String prefix) {
+      this(Executors.defaultThreadFactory(), prefix);
+    }
+
+    private NamedThreadFactory(ThreadFactory backingThreadFactory, String prefix) {
+      this.backingThreadFactory = backingThreadFactory;
+      this.prefix = prefix;
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+      Thread thread = this.backingThreadFactory.newThread(r);
+      thread.setName(prefix + count.getAndIncrement());
+      return thread;
+    }
+  }
 }

src/test/resources/logback-test.xml

@@ -8,7 +8,7 @@
     <logger name="com.rabbitmq.client.AmqpClientTestExtension" level="debug" />
     <logger name="com.rabbitmq.client.test.server.TopicPermissions" level="debug" />
     <logger name="com.rabbitmq.client.impl.AbstractMetricsCollector" level="debug" />
-    <logger name="com.rabbitmq.client.impl.NettyFrameHandlerFactory" level="warn" />
+    <logger name="com.rabbitmq.client.impl.NettyFrameHandlerFactory" level="debug" />
 
     <root level="warn">
         <appender-ref ref="STDOUT" />