[Good First Issue] Migrate to Monitored Thread Pools by Replacing Native ExecutorService

[Chillax-0v0]

Problem Description

Java's native Executors.new*ThreadPool() methods lack visibility into thread pool workloads. Missing queue size monitoring makes it hard to:

• Detect thread pool congestion (e.g., blocking I/O tasks)
• Diagnose "silent hangs" during incidents
• Optimize pool sizing

Our utility class com.automq.stream.utils.Threads provides monitored alternatives (e.g., newFixedThreadPoolWithMonitor) that log queue sizes to s3stream-threads.log. However, AutoMQ code still contains unsafe native pool initializations.

Required Changes

1. Replace all native pool creations (Executors.newFixedThreadPool, newCachedThreadPool, etc.) in AutoMQ code with matching Threads methods
2. Extend Threads class if needed:
   • Add overloaded methods for parameter variations (e.g., custom thread factories)
   • Keep method signatures compatible with native Executors patterns
3. Ensure no changes are made to original Apache Kafka code paths

Scope Guidance

Focus on code in these patterns:

• Located in com.automq.* packages
• Clearly identified as AutoMQ-specific additions (search for "AutoMQ inject" and "AutoMQ for Kafka inject")
• Files with the AutoMQ HK Limited header

[Akika404]

/assign

[HrithikPadavala]

Hi @Chillax-0v0, please assign me.

[Akika404]

Hi @Chillax-0v0 , hmmm...maybe I need some help. The code contains some method likes Executors.newScheduledThreadPool(), but the class Threads and S3StreamThreadPoolMonitor do not have corresponding ScheduledExecutorService methods. Also, it seems that Threads#newSingleThreadScheduledExecutor is not monitored.

Should I add the corresponding method in S3StreamThreadPoolMonitor and Threads, or just replace them with Threads.newSingleThreadScheduledExecutor()?🤔

[Chillax-0v0]

Hi, @ciimina,

Thank you for raising this question. The monitoring implementation for ScheduledExecutorService is actually tracked in a separate issue (#2357). The current scope of work focuses only on replacing standard thread pools created via Executors.new*ThreadPool(), not scheduled executors.

Regarding your specific case: since Threads.newSingleThreadScheduledExecutor() currently lacks monitoring capabilities, no action is required for these instances under the current issue. The design intent for scheduled executors assumes they won’t handle heavy workloads where queue monitoring becomes critical.

Please proceed by limiting changes to non-scheduled thread pools as outlined in the original problem description. Scheduled executor monitoring will be addressed in its dedicated follow-up task. We appreciate your attention to detail in identifying this edge case.

[LeedY1232]

/assign

[shawngao-org]

/assign