Log thread dump in case of long-running test

Author: acogoluegnes

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

@@ -16,6 +16,8 @@
 
 package com.rabbitmq.client;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.jupiter.api.extension.ConditionEvaluationResult.disabled;
 import static org.junit.jupiter.api.extension.ConditionEvaluationResult.enabled;
@@ -28,11 +30,29 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.nio.NioIoHandler;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.lang.management.LockInfo;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MonitorInfo;
+import java.lang.management.RuntimeMXBean;
+import java.lang.management.ThreadInfo;
 import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.AfterEachCallback;
@@ -59,6 +79,8 @@ public class AmqpClientTestExtension
   private static final ExtensionContext.Namespace NAMESPACE =
       ExtensionContext.Namespace.create(AmqpClientTestExtension.class);
 
+  private static final AtomicReference<String> CURRENT_TEST = new AtomicReference<>();
+
   private static ExtensionContext.Store store(ExtensionContext extensionContext) {
     return extensionContext.getRoot().getStore(NAMESPACE);
   }
@@ -108,6 +130,16 @@ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext con
   @Override
   public void beforeAll(ExtensionContext context) {
     if (TestUtils.isNetty()) {
+      Duration timeout = Duration.ofMinutes(20);
+      ScheduledFuture<?> task = executor(context).schedule(() -> {
+        try {
+          LOGGER.warn("Test {} has been running for {}", CURRENT_TEST.get(), timeout);
+          logThreadDump();
+        } catch (Exception e) {
+          LOGGER.warn("Error during test timeout task", e);
+        }
+      }, timeout.toMillis(), MILLISECONDS);
+      store(context).put("threadDumpTask", task);
       ThreadFactory tf = new NamedThreadFactory(context.getTestClass().get().getSimpleName() + "-");
       EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(tf, NioIoHandler.newFactory());
       store(context)
@@ -118,11 +150,10 @@ public void beforeAll(ExtensionContext context) {
 
   @Override
   public void beforeEach(ExtensionContext context) {
-    LOGGER.info(
-        "Starting test: {}.{} (IO layer: {})",
-        context.getTestClass().get().getSimpleName(),
-        context.getTestMethod().get().getName(),
-        TestUtils.IO_LAYER);
+    String test = String.format("%s.%s", context.getTestClass().get().getSimpleName(),
+        context.getTestMethod().get().getName());
+    CURRENT_TEST.set(test);
+    LOGGER.info("Starting test: {} (IO layer: {})", test, TestUtils.IO_LAYER);
   }
 
   @Override
@@ -137,34 +168,44 @@ public void afterEach(ExtensionContext context) {
   public void afterAll(ExtensionContext context) {
     if (TestUtils.isNetty()) {
       TestUtils.resetEventLoopGroup();
+      ScheduledFuture<?> threadDumpTask = store(context).get("threadDumpTask", ScheduledFuture.class);
+      if (threadDumpTask != null) {
+        threadDumpTask.cancel(true);
+      }
       EventLoopGroup eventLoopGroup = eventLoopGroup(context);
-      ExecutorServiceCloseableResourceWrapper wrapper =
-          context
-              .getRoot()
-              .getStore(ExtensionContext.Namespace.GLOBAL)
-              .getOrComputeIfAbsent(ExecutorServiceCloseableResourceWrapper.class);
-
-      wrapper
-          .executorService
-          .submit(
-              () -> {
-                try {
-                  eventLoopGroup.shutdownGracefully(0, 0, SECONDS).get(10, SECONDS);
-                } catch (InterruptedException e) {
-                  Thread.currentThread().interrupt();
-                } catch (Exception e) {
-                  LOGGER.warn("Error while asynchronously closing Netty event loop group", e);
-                }
-              });
+      ExecutorService executor = executor(context);
+
+      executor.submit(() -> {
+        try {
+          eventLoopGroup.shutdownGracefully(0, 0, SECONDS).get(10, SECONDS);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        } catch (Exception e) {
+          LOGGER.warn("Error while asynchronously closing Netty event loop group", e);
+        }
+      });
     }
   }
 
+  private static ScheduledExecutorService executor(ExtensionContext context) {
+    ExecutorServiceCloseableResourceWrapper wrapper =
+        context
+            .getRoot()
+            .getStore(ExtensionContext.Namespace.GLOBAL)
+            .getOrComputeIfAbsent(ExecutorServiceCloseableResourceWrapper.class);
+    return wrapper.executor();
+  }
+
   private static class ExecutorServiceCloseableResourceWrapper implements AutoCloseable {
 
-    private final ExecutorService executorService;
+    private final ScheduledExecutorService executorService;
 
     private ExecutorServiceCloseableResourceWrapper() {
-      this.executorService = Executors.newCachedThreadPool();
+      this.executorService = Executors.newScheduledThreadPool(2);
+    }
+
+    private ScheduledExecutorService executor() {
+      return this.executorService;
     }
 
     @Override
@@ -197,4 +238,109 @@ public Thread newThread(Runnable r) {
       return thread;
     }
   }
+
+  private static void logThreadDump() {
+    PlainTextThreadDumpFormatter formatter = new PlainTextThreadDumpFormatter();
+    ThreadInfo[] threadInfos =
+        ManagementFactory.getThreadMXBean().dumpAllThreads(true, true);
+    String threadDump = formatter.format(threadInfos);
+    LOGGER.warn(threadDump);
+  }
+
+  // from Spring Boot's PlainTextThreadDumpFormatter
+  private static class PlainTextThreadDumpFormatter {
+
+    String format(ThreadInfo[] threads) {
+      StringWriter dump = new StringWriter();
+      PrintWriter writer = new PrintWriter(dump);
+      writePreamble(writer);
+      for (ThreadInfo info : threads) {
+        writeThread(writer, info);
+      }
+      return dump.toString();
+    }
+
+    private void writePreamble(PrintWriter writer) {
+      DateTimeFormatter dateFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+      writer.println(dateFormat.format(LocalDateTime.now()));
+      RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
+      writer.printf(
+          "Full thread dump %s (%s %s):%n",
+          runtime.getVmName(), runtime.getVmVersion(), System.getProperty("java.vm.info"));
+      writer.println();
+    }
+
+    private void writeThread(PrintWriter writer, ThreadInfo info) {
+      writer.printf("\"%s\" - Thread t@%d%n", info.getThreadName(), info.getThreadId());
+      writer.printf("   %s: %s%n", Thread.State.class.getCanonicalName(), info.getThreadState());
+      writeStackTrace(writer, info, info.getLockedMonitors());
+      writer.println();
+      writeLockedOwnableSynchronizers(writer, info);
+      writer.println();
+    }
+
+    private void writeStackTrace(
+        PrintWriter writer, ThreadInfo info, MonitorInfo[] lockedMonitors) {
+      int depth = 0;
+      for (StackTraceElement element : info.getStackTrace()) {
+        writeStackTraceElement(
+            writer, element, info, lockedMonitorsForDepth(lockedMonitors, depth), depth == 0);
+        depth++;
+      }
+    }
+
+    private List<MonitorInfo> lockedMonitorsForDepth(MonitorInfo[] lockedMonitors, int depth) {
+      return Stream.of(lockedMonitors)
+          .filter((lockedMonitor) -> lockedMonitor.getLockedStackDepth() == depth)
+          .collect(Collectors.toList());
+    }
+
+    private void writeStackTraceElement(
+        PrintWriter writer,
+        StackTraceElement element,
+        ThreadInfo info,
+        List<MonitorInfo> lockedMonitors,
+        boolean firstElement) {
+      writer.printf("\tat %s%n", element.toString());
+      LockInfo lockInfo = info.getLockInfo();
+      if (firstElement && lockInfo != null) {
+        if (element.getClassName().equals(Object.class.getName())
+            && element.getMethodName().equals("wait")) {
+          writer.printf("\t- waiting on %s%n", format(lockInfo));
+        } else {
+          String lockOwner = info.getLockOwnerName();
+          if (lockOwner != null) {
+            writer.printf(
+                "\t- waiting to lock %s owned by \"%s\" t@%d%n",
+                format(lockInfo), lockOwner, info.getLockOwnerId());
+          } else {
+            writer.printf("\t- parking to wait for %s%n", format(lockInfo));
+          }
+        }
+      }
+      writeMonitors(writer, lockedMonitors);
+    }
+
+    private String format(LockInfo lockInfo) {
+      return String.format("<%x> (a %s)", lockInfo.getIdentityHashCode(), lockInfo.getClassName());
+    }
+
+    private void writeMonitors(PrintWriter writer, List<MonitorInfo> lockedMonitorsAtCurrentDepth) {
+      for (MonitorInfo lockedMonitor : lockedMonitorsAtCurrentDepth) {
+        writer.printf("\t- locked %s%n", format(lockedMonitor));
+      }
+    }
+
+    private void writeLockedOwnableSynchronizers(PrintWriter writer, ThreadInfo info) {
+      writer.println("   Locked ownable synchronizers:");
+      LockInfo[] lockedSynchronizers = info.getLockedSynchronizers();
+      if (lockedSynchronizers == null || lockedSynchronizers.length == 0) {
+        writer.println("\t- None");
+      } else {
+        for (LockInfo lockedSynchronizer : lockedSynchronizers) {
+          writer.printf("\t- Locked %s%n", format(lockedSynchronizer));
+        }
+      }
+    }
+  }
 }