perf: Replace Unsafe ScheduledExecutorService Initialization with Exception-Handling Wrapper (#2414)

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

* perf: pr-issue-2357

Author: LeedY1232

Diff for: core/src/main/java/kafka/autobalancer/LoadRetriever.java

@@ -52,6 +52,7 @@
 import org.apache.kafka.server.config.QuotaConfigs;
 
 import com.automq.stream.utils.LogContext;
+import com.automq.stream.utils.Threads;
 
 import java.time.Duration;
 import java.util.Collections;
@@ -65,7 +66,6 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
@@ -105,7 +105,9 @@ public LoadRetriever(AutoBalancerControllerConfig config, Controller controller,
         this.bootstrapServerMapInUse = new HashMap<>();
         this.lock = new ReentrantLock();
         this.cond = lock.newCondition();
-        this.mainExecutorService = Executors.newSingleThreadScheduledExecutor(new AutoBalancerThreadFactory("load-retriever-main"));
+        this.mainExecutorService =
+            Threads.newSingleThreadScheduledExecutor(
+                new AutoBalancerThreadFactory("load-retriever-main"), logger);
         leaderEpochInitialized = false;
         staticConfig = new StaticAutoBalancerConfig(config.originals(), false);
         listenerName = staticConfig.getString(StaticAutoBalancerConfig.AUTO_BALANCER_CLIENT_LISTENER_NAME_CONFIG);

Diff for: core/src/main/scala/kafka/log/stream/s3/metadata/StreamMetadataManager.java

@@ -39,6 +39,7 @@
 import com.automq.stream.s3.operator.ObjectStorage.ReadOptions;
 import com.automq.stream.s3.streams.StreamMetadataListener;
 import com.automq.stream.utils.FutureUtil;
+import com.automq.stream.utils.Threads;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -52,7 +53,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 import io.netty.buffer.ByteBuf;
@@ -77,7 +77,8 @@ public StreamMetadataManager(BrokerServer broker, int nodeId, ObjectReaderFactor
         this.pendingGetObjectsTasks = new LinkedList<>();
         this.objectReaderFactory = objectReaderFactory;
         this.indexCache = indexCache;
-        this.pendingExecutorService = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pending-get-objects-task-executor"));
+        this.pendingExecutorService =
+            Threads.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pending-get-objects-task-executor"), LOGGER);
         broker.metadataLoader().installPublishers(List.of(this)).join();
     }
 

Diff for: core/src/main/scala/kafka/log/stream/s3/network/ControllerRequestSender.java

@@ -23,6 +23,8 @@
 import org.apache.kafka.server.ControllerRequestCompletionHandler;
 import org.apache.kafka.server.NodeToControllerChannelManager;
 
+import com.automq.stream.utils.Threads;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,18 +33,18 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
-import io.netty.util.concurrent.DefaultThreadFactory;
-
 public class ControllerRequestSender {
+
     private static final Logger LOGGER = LoggerFactory.getLogger(ControllerRequestSender.class);
+
     private static final long MAX_RETRY_DELAY_MS = 10 * 1000; // 10s
+
     private final RetryPolicyContext retryPolicyContext;
 
     private final NodeToControllerChannelManager channelManager;
@@ -55,7 +57,8 @@ public ControllerRequestSender(BrokerServer brokerServer, RetryPolicyContext ret
         this.retryPolicyContext = retryPolicyContext;
         this.channelManager = brokerServer.newNodeToControllerChannelManager("s3stream-to-controller", 60000);
         this.channelManager.start();
-        this.retryService = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("controller-request-retry-sender"));
+        this.retryService =
+            Threads.newSingleThreadScheduledExecutor("controller-request-retry-sender", false, LOGGER);
         this.requestAccumulatorMap = new ConcurrentHashMap<>();
     }
 

Diff for: core/src/main/scala/kafka/log/streamaspect/ElasticUnifiedLog.scala

@@ -12,7 +12,7 @@
 package kafka.log.streamaspect
 
 import com.automq.stream.api.Client
-import com.automq.stream.utils.FutureUtil
+import com.automq.stream.utils.{FutureUtil, Threads}
 import kafka.cluster.PartitionSnapshot
 import kafka.log._
 import kafka.log.streamaspect.ElasticUnifiedLog.{CheckpointExecutor, MaxCheckpointIntervalBytes, MinCheckpointIntervalMs}
@@ -21,7 +21,7 @@ import kafka.utils.Logging
 import org.apache.kafka.common.errors.OffsetOutOfRangeException
 import org.apache.kafka.common.errors.s3.StreamFencedException
 import org.apache.kafka.common.record.{MemoryRecords, RecordVersion}
-import org.apache.kafka.common.utils.{ThreadUtils, Time}
+import org.apache.kafka.common.utils.Time
 import org.apache.kafka.common.{TopicPartition, Uuid}
 import org.apache.kafka.server.common.{MetadataVersion, OffsetAndEpoch}
 import org.apache.kafka.server.util.Scheduler
@@ -33,7 +33,7 @@ import java.nio.ByteBuffer
 import java.nio.file.Path
 import java.util
 import java.util.concurrent.atomic.LongAdder
-import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, CopyOnWriteArrayList, Executors}
+import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, CopyOnWriteArrayList}
 import scala.jdk.CollectionConverters.CollectionHasAsScala
 import scala.util.{Failure, Success, Try}
 
@@ -291,8 +291,10 @@ class ElasticUnifiedLog(_logStartOffset: Long,
 }
 
 object ElasticUnifiedLog extends Logging {
-    private val CheckpointExecutor = Executors.newSingleThreadScheduledExecutor(ThreadUtils.createThreadFactory("checkpoint-executor", true))
-    private val MaxCheckpointIntervalBytes = 50 * 1024 * 1024
+    private val CheckpointExecutor = {
+      Threads.newSingleThreadScheduledExecutor("checkpoint-executor", true, logger.underlying)
+    }
+  private val MaxCheckpointIntervalBytes = 50 * 1024 * 1024
     private val MinCheckpointIntervalMs = 10 * 1000
     private val Logs = new ConcurrentHashMap[TopicPartition, ElasticUnifiedLog]()
     // fuzzy dirty bytes for checkpoint, it's ok not thread safe

Diff for: s3stream/src/main/java/com/automq/stream/s3/index/LocalStreamRangeIndexCache.java

@@ -22,7 +22,7 @@
 import com.automq.stream.s3.operator.ObjectStorage;
 import com.automq.stream.s3.operator.ObjectStorage.ReadOptions;
 import com.automq.stream.utils.Systems;
-import com.automq.stream.utils.ThreadUtils;
+import com.automq.stream.utils.Threads;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,7 +40,6 @@
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -62,8 +61,8 @@ public class LocalStreamRangeIndexCache implements S3StreamClient.StreamLifeCycl
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
     private final Lock readLock = lock.readLock();
     private final Lock writeLock = lock.writeLock();
-    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(
-        ThreadUtils.createThreadFactory("upload-index", true));
+    private final ScheduledExecutorService executorService =
+        Threads.newSingleThreadScheduledExecutor("upload-index", true, LOGGER);
     private final Queue<CompletableFuture<Void>> uploadQueue = new LinkedList<>();
     private final CompletableFuture<Void> initCf = new CompletableFuture<>();
     private final AtomicBoolean pruned = new AtomicBoolean(false);

Diff for: s3stream/src/main/java/com/automq/stream/s3/network/AsyncNetworkBandwidthLimiter.java

@@ -15,6 +15,7 @@
 import com.automq.stream.s3.metrics.S3StreamMetricsManager;
 import com.automq.stream.s3.metrics.stats.NetworkStats;
 import com.automq.stream.utils.LogContext;
+import com.automq.stream.utils.Threads;
 
 import org.slf4j.Logger;
 
@@ -56,7 +57,8 @@ public AsyncNetworkBandwidthLimiter(Type type, long tokenSize, int refillInterva
         this.availableTokens = this.tokenSize;
         this.maxTokens = maxTokens;
         this.queuedCallbacks = new PriorityQueue<>();
-        this.refillThreadPool = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("refill-bucket-thread"));
+        this.refillThreadPool =
+            Threads.newSingleThreadScheduledExecutor(new DefaultThreadFactory("refill-bucket-thread"), LOGGER);
         this.callbackThreadPool = Executors.newFixedThreadPool(1, new DefaultThreadFactory("callback-thread"));
         this.callbackThreadPool.execute(this::run);
         this.refillThreadPool.scheduleAtFixedRate(this::refillToken, refillIntervalMs, refillIntervalMs, TimeUnit.MILLISECONDS);

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

@@ -12,6 +12,7 @@
 package com.automq.stream.utils.threads;
 
 import com.automq.stream.utils.ThreadUtils;
+import com.automq.stream.utils.Threads;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,7 +21,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -30,9 +30,9 @@
 
 public class S3StreamThreadPoolMonitor {
     private static final List<ThreadPoolWrapper> MONITOR_EXECUTOR = new CopyOnWriteArrayList<>();
-    private static final ScheduledExecutorService MONITOR_SCHEDULED = Executors.newSingleThreadScheduledExecutor(
-        ThreadUtils.createThreadFactory("ThreadPoolMonitor-%d", true));
     private static Logger waterMarkLogger = LoggerFactory.getLogger(S3StreamThreadPoolMonitor.class);
+    private static final ScheduledExecutorService MONITOR_SCHEDULED =
+        Threads.newSingleThreadScheduledExecutor(ThreadUtils.createThreadFactory("ThreadPoolMonitor-%d", true), waterMarkLogger);
     private static volatile long threadPoolStatusPeriodTime = TimeUnit.SECONDS.toMillis(3);
 
     public static void config(Logger waterMarkLoggerConfig, long threadPoolStatusPeriodTimeConfig) {

Diff for: tools/src/main/java/org/apache/kafka/tools/automq/perf/ProducerService.java

@@ -22,6 +22,8 @@
 import org.apache.kafka.common.utils.ThreadUtils;
 import org.apache.kafka.tools.automq.perf.TopicService.Topic;
 
+import com.automq.stream.utils.Threads;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,7 +58,8 @@ public class ProducerService implements AutoCloseable {
     private static final Logger LOGGER = LoggerFactory.getLogger(ProducerService.class);
 
     private final List<Producer> producers = new LinkedList<>();
-    private final ScheduledExecutorService adjustRateExecutor = Executors.newSingleThreadScheduledExecutor(ThreadUtils.createThreadFactory("perf-producer-rate-adjust", true));
+    private final ScheduledExecutorService adjustRateExecutor =
+        Threads.newSingleThreadScheduledExecutor("perf-producer-rate-adjust", true, LOGGER);
     private final ExecutorService executor = Executors.newCachedThreadPool(ThreadUtils.createThreadFactory("perf-producer", false));
 
     /**