fix checkstyle

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGeneratorUtils.java

@@ -23,6 +23,8 @@
 import io.prometheus.client.CollectorRegistry;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
 import java.util.ArrayList;
 import java.util.Enumeration;
 import java.util.HashMap;
@@ -76,7 +78,7 @@ public static void generateSystemMetrics(SimpleTextOutputStream stream, String c
                 if (!sample.labelNames.contains("cluster")) {
                     stream.write("cluster=\"").write(cluster).write('"');
                     // If label is empty, should not append ','.
-                    if (!CollectionUtils.isEmpty(sample.labelNames)){
+                    if (!CollectionUtils.isEmpty(sample.labelNames)) {
                         stream.write(",");
                     }
                 }
@@ -101,11 +103,19 @@ public static void generateSystemMetrics(SimpleTextOutputStream stream, String c
         }
     }
 
+    public static final String THREAD_BLOCK_COUNTS = "thread_blocked_count";
+    public static final String THREAD_BLOCK_TIME_MS = "thread_blocked_time_ms";
+    public static final String THREAD_WAIT_COUNTS = "thread_waited_counts";
+    public static final String THREAD_WAIT_TIME_MS = "thread_waited_time_ms";
+
     public static List<Metrics> generateThreadPoolMonitorMetrics(String cluster) {
         List<Metrics> metrics = new ArrayList<>();
 
-        Map<Long, Long> threadStat = new HashMap<>(ThreadMonitor.THREAD_LAST_ACTIVE_TIMESTAMP);
-        Map<Long, String> threadIdMapping = new HashMap<>(ThreadMonitor.THREAD_ID_TO_NAME);
+        Map<Long, Long> threadStat = new HashMap<>(ThreadMonitor.THREAD_LAST_ACTIVE_TIMESTAMP.size());
+        threadStat.putAll(ThreadMonitor.THREAD_LAST_ACTIVE_TIMESTAMP);
+
+        Map<Long, String> threadIdMapping = new HashMap<>(ThreadMonitor.THREAD_ID_TO_NAME.size());
+        threadIdMapping.putAll(ThreadMonitor.THREAD_ID_TO_NAME);
 
         threadStat.forEach((tid, lastActiveTimestamp) -> {
             Map<String, String> dimensionMap = new HashMap<>();
@@ -133,6 +143,28 @@ public static List<Metrics> generateThreadPoolMonitorMetrics(String cluster) {
 
         metrics.add(m);
 
+        ThreadMXBean threadMXBean = ThreadPoolMonitor.THREAD_MX_BEAN;
+        if (threadMXBean.isThreadContentionMonitoringSupported()) {
+            long[] allThreadIds = threadMXBean.getAllThreadIds();
+            ThreadInfo[] threadInfo = threadMXBean.getThreadInfo(allThreadIds, 0);
+            for (ThreadInfo info : threadInfo) {
+                if (info != null) {
+                    Map<String, String> dimensions = new HashMap<>();
+                    dimensions.put("cluster", cluster);
+                    dimensions.put("threadName", info.getThreadName());
+                    dimensions.put("tid", String.valueOf(info.getThreadId()));
+
+                    Metrics threadMetric = Metrics.create(dimensions);
+                    threadMetric.put(THREAD_BLOCK_COUNTS, info.getBlockedCount());
+                    threadMetric.put(THREAD_BLOCK_TIME_MS, info.getBlockedTime());
+                    threadMetric.put(THREAD_WAIT_COUNTS, info.getWaitedCount());
+                    threadMetric.put(THREAD_WAIT_TIME_MS, info.getWaitedTime());
+
+                    metrics.add(threadMetric);
+                }
+            }
+        }
+
         return metrics;
     }
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/WebExecutorThreadPool.java

@@ -18,12 +18,14 @@
  */
 package org.apache.pulsar.broker.web;
 
-import io.netty.util.concurrent.DefaultThreadFactory;
-import io.netty.util.concurrent.FastThreadLocalThread;
+
+import java.lang.reflect.Field;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.common.util.ExecutorProvider;
 import org.apache.pulsar.common.util.ThreadMonitor;
+import org.apache.pulsar.common.util.ThreadPoolMonitor;
 import org.eclipse.jetty.util.BlockingArrayQueue;
 import org.eclipse.jetty.util.thread.ExecutorThreadPool;
 
@@ -48,22 +50,25 @@ protected void beforeExecute(Thread t, Runnable r) {
             protected void afterExecute(Runnable r, Throwable t) {
                 ThreadMonitor.refreshThreadState(Thread.currentThread());
             }
-        }, Math.min(8, maxThreads));
+        }, -1);
 
-        this.threadFactory = new DefaultThreadFactory(namePrefix) {
-            @Override
-            protected Thread newThread(Runnable r, String name) {
-                return new FastThreadLocalThread(threadGroup, r, name) {
-                    @Override
-                    public void run() {
-                        super.run();
+        this.threadFactory = new ExecutorProvider.ExtendedThreadFactory(namePrefix);
+        tryMonitorThreadExecutor(maxThreads);
+    }
 
-                        // execute when thread exit.
-                        ThreadMonitor.remove(Thread.currentThread());
-                    }
-                };
-            }
-        };
+    private void tryMonitorThreadExecutor(int threadNumber) {
+        ThreadPoolExecutor executor;
+        try {
+            Field executorField = this.getClass().getSuperclass().getDeclaredField("_executor");
+            executorField.setAccessible(true);
+            executor = (ThreadPoolExecutor) executorField.get(this);
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            executor = null;
+        }
+
+        if (executor != null) {
+            ThreadPoolMonitor.registerMultiThreadExecutor(executor, threadNumber);
+        }
     }
 
     @Override