chore(zerozone2): schedule logging proxy mapping (#2958)

Signed-off-by: Robin Han <hanxvdovehx@gmail.com>

Author: superhx

core/src/main/java/kafka/automq/zerozone/ProxyNodeMapping.java

@@ -28,6 +28,8 @@
 import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.metadata.BrokerRegistration;
 
+import com.automq.stream.utils.Threads;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,6 +41,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import software.amazon.awssdk.annotations.NotNull;
@@ -57,7 +60,7 @@ class ProxyNodeMapping {
     private final MetadataCache metadataCache;
     private final List<ProxyTopologyChangeListener> listeners = new CopyOnWriteArrayList<>();
 
-    volatile Map<String, Map<Integer, BrokerRegistration>> main2proxyByRack = new HashMap<>();
+    volatile Map<String /* proxy rack */, Map<Integer /* main nodeId */, BrokerRegistration /* proxy */>> main2proxyByRack = new HashMap<>();
     volatile boolean inited = false;
 
     public ProxyNodeMapping(Node currentNode, String currentRack, String interBrokerListenerName,
@@ -66,6 +69,7 @@ public ProxyNodeMapping(Node currentNode, String currentRack, String interBroker
         this.currentNode = currentNode;
         this.currentRack = currentRack;
         this.metadataCache = metadataCache;
+        Threads.COMMON_SCHEDULER.scheduleWithFixedDelay(() -> logMapping(main2proxyByRack), 1, 1, TimeUnit.MINUTES);
     }
 
     /**
@@ -215,6 +219,7 @@ public void onChange(MetadataDelta delta, MetadataImage image) {
             });
         });
         this.main2proxyByRack = calMain2proxyByRack(rack2brokers);
+        logMapping(main2proxyByRack);
         notifyListeners(this.main2proxyByRack);
     }
 
@@ -354,6 +359,18 @@ static void tryFreeController(List<ProxyNode> proxyNodes, double avg) {
         }
     }
 
+
+
+    static void logMapping(Map<String, Map<Integer, BrokerRegistration>> main2proxyByRack) {
+        StringBuilder sb = new StringBuilder();
+        main2proxyByRack.forEach((rack, main2proxy) ->
+            main2proxy.forEach((mainNodeId, proxyNode) ->
+                sb.append(" Main ").append(mainNodeId).append(" => Proxy ").append(proxyNode.id()).append("(").append(rack).append(")\n")
+            )
+        );
+        LOGGER.info("ProxyNodeMapping:\n{}", sb);
+    }
+
     static class ProxyNode implements thirdparty.com.github.jaskey.consistenthash.Node, Comparable<ProxyNode> {
         final BrokerRegistration node;
         final List<Integer> mainNodeIds = new ArrayList<>();

s3stream/src/main/java/com/automq/stream/utils/threads/S3StreamThreadPoolMonitor.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
 public class S3StreamThreadPoolMonitor {
@@ -97,6 +98,7 @@ public static ThreadPoolExecutor createAndMonitor(int corePoolSize,
                 ThreadUtils.createFastThreadLocalThreadFactory(name + "-%d", isDaemon) :
                 ThreadUtils.createThreadFactory(name + "-%d", isDaemon);
 
+        AtomicReference<ThreadPoolWrapper> wrapperRef = new AtomicReference<>();
         ThreadPoolExecutor executor = new ThreadPoolExecutor(
             corePoolSize,
             maximumPoolSize,
@@ -110,16 +112,23 @@ protected void afterExecute(Runnable r, Throwable t) {
                 super.afterExecute(r, t);
                 afterExecutionHook.apply(t);
             }
+
+            @Override
+            protected void terminated() {
+                super.terminated();
+                ThreadPoolWrapper wrapper = wrapperRef.get();
+                if (wrapper != null) {
+                    MONITOR_EXECUTOR.remove(wrapper);
+                }
+            }
         };
         List<ThreadPoolStatusMonitor> printers = new ArrayList<>();
         printers.add(new ThreadPoolQueueSizeMonitor(queueCapacity));
         printers.addAll(threadPoolStatusMonitors);
 
-        MONITOR_EXECUTOR.add(ThreadPoolWrapper.builder()
-            .name(name)
-            .threadPoolExecutor(executor)
-            .statusPrinters(printers)
-            .build());
+        ThreadPoolWrapper wrapper = ThreadPoolWrapper.builder().name(name).threadPoolExecutor(executor).statusPrinters(printers).build();
+        wrapperRef.set(wrapper);
+        MONITOR_EXECUTOR.add(wrapper);
         return executor;
     }