[ISSUE #9699] Optimize shutdown process and resource management (#9700)

* Optimize shutdown process and resource management

- Improve BrokerController shutdown flow for graceful shutdown
- Optimize BrokerStartup startup and shutdown logic
- Enhance ClientHousekeepingService resource cleanup
- Improve shutdown handling for various processors
- Optimize resource management in storage layer components
- Enhance lifecycle management for statistics manager
- Improve shutdown flow for timer components

* Fix this.popMessageProcessor.getPopLongPollingService() not shutdown

* Fix test shutdown state transition issue

- Add proper null checks and exception handling in test cleanup
- Prevent IllegalStateException during test teardown
- Ensure graceful test cleanup without state conflicts

* Fix DefaultMessageStoreCleanFilesTest can not pass

* Fix CombineConsumeQueueStoreTest can not pass

* Polish the code

* Polish the code

* Ignore flaky test first

Author: RongtongJin

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -1460,6 +1460,18 @@ protected void shutdownBasicService() {
             this.transactionalMessageService.close();
         }
 
+        if (this.transactionalMessageCheckListener != null) {
+            this.transactionalMessageCheckListener.shutdown();
+        }
+
+        if (transactionalMessageCheckService != null) {
+            this.transactionalMessageCheckService.shutdown();
+        }
+
+        if (transactionMetricsFlushService != null) {
+            this.transactionMetricsFlushService.shutdown();
+        }
+
         if (this.notificationProcessor != null) {
             this.notificationProcessor.getPopLongPollingService().shutdown();
         }
@@ -1483,10 +1495,6 @@ protected void shutdownBasicService() {
             this.broadcastOffsetManager.shutdown();
         }
 
-        if (this.messageStore != null) {
-            this.messageStore.shutdown();
-        }
-
         if (this.replicasManager != null) {
             this.replicasManager.shutdown();
         }
@@ -1626,6 +1634,10 @@ protected void shutdownBasicService() {
                 brokerAttachedPlugin.shutdown();
             }
         }
+
+        if (this.messageStore != null) {
+            this.messageStore.shutdown();
+        }
     }
 
     public void shutdown() {

broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java

@@ -87,13 +87,14 @@ public static ConfigContext parseCmdLine(String[] args) throws Exception {
             System.exit(-1);
         }
 
-        ConfigContext configContext = null;
-        String filePath;
+        ConfigContext configContext;
+        String filePath = null;
         if (commandLine.hasOption('c')) {
             filePath = commandLine.getOptionValue('c');
-            configContext = configFileToConfigContext(filePath);
         }
 
+        configContext = configFileToConfigContext(filePath);
+
         if (commandLine.hasOption('p') && configContext != null) {
             Logger console = LoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
             MixAll.printObjectProperties(console, configContext.getBrokerConfig());

broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java

@@ -41,14 +41,11 @@ public ClientHousekeepingService(final BrokerController brokerController) {
 
     public void start() {
 
-        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    ClientHousekeepingService.this.scanExceptionChannel();
-                } catch (Throwable e) {
-                    log.error("Error occurred when scan not active client channels.", e);
-                }
+        this.scheduledExecutorService.scheduleAtFixedRate(() -> {
+            try {
+                ClientHousekeepingService.this.scanExceptionChannel();
+            } catch (Throwable e) {
+                log.error("Error occurred when scan not active client channels.", e);
             }
         }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
     }

broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java

@@ -74,6 +74,12 @@ public PopReviveService[] getPopReviveServices() {
         return popReviveServices;
     }
 
+    public void shutdown() throws Exception {
+        for (PopReviveService popReviveService : popReviveServices) {
+            popReviveService.shutdown();
+        }
+    }
+
     public void startPopReviveService() {
         for (PopReviveService popReviveService : popReviveServices) {
             popReviveService.start();

broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java

@@ -55,6 +55,10 @@ public NotificationProcessor(final BrokerController brokerController) {
         this.popLongPollingService = new PopLongPollingService(brokerController, this, true);
     }
 
+    public void shutdown() throws Exception {
+        this.popLongPollingService.shutdown();
+    }
+
     @Override
     public boolean rejectRequest() {
         return false;

broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java

@@ -126,8 +126,10 @@ public void run() {
         if (!isShouldRunning()) {
             return;
         }
-        while (this.buffer.size() > 0 || getOffsetTotalSize() > 0) {
-            scan();
+        if (!brokerController.getBrokerConfig().isInBrokerContainer()) {
+            while (this.buffer.size() > 0 || getOffsetTotalSize() > 0) {
+                scan();
+            }
         }
     }
 

broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java

@@ -123,6 +123,12 @@ public PopMessageProcessor(final BrokerController brokerController) {
         this.ckMessageNumber = new AtomicLong();
     }
 
+    public void shutdown() throws Exception {
+        popLongPollingService.shutdown();
+        queueLockManager.shutdown();
+        popBufferMergeService.shutdown();
+    }
+
     protected String getReviveTopic() {
         return reviveTopic;
     }

broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java

@@ -90,7 +90,7 @@ public BrokerController getBrokerController() {
         return brokerController;
     }
 
-    public void shutDown() {
+    public void shutdown() {
         if (executorService != null) {
             executorService.shutdown();
         }

common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java

@@ -487,10 +487,16 @@ public synchronized boolean shutdown() {
                 return true;
             }
 
+            manualCompactionThread.shutdownNow();
+
+            manualCompactionThread.awaitTermination(30, TimeUnit.SECONDS);
+
             final FlushOptions flushOptions = new FlushOptions();
             flushOptions.setWaitForFlush(true);
             try {
                 flush(flushOptions);
+            } catch (Throwable e) {
+                LOGGER.error("flush rocksdb wal failed when shutdown", e);
             } finally {
                 flushOptions.close();
             }
@@ -521,10 +527,22 @@ public synchronized boolean shutdown() {
             }
             //4. close db.
             if (db != null && !this.readOnly) {
-                this.db.syncWal();
+                try {
+                    this.db.syncWal();
+                } catch (Throwable e) {
+                    LOGGER.error("rocksdb sync wal failed when shutdown", e);
+                } finally {
+                    flushOptions.close();
+                }
+
             }
             if (db != null) {
-                this.db.closeE();
+                try {
+                    this.db.closeE();
+                } catch (Throwable e) {
+                    LOGGER.error("rocksdb db closeE failed when shutdown", e);
+                }
+
             }
             // Close DBOptions after RocksDB instance is closed.
             if (this.options != null) {

common/src/main/java/org/apache/rocketmq/common/statistics/StatisticsManager.java

@@ -154,4 +154,8 @@ public StatisticsItemStateGetter getStatisticsItemStateGetter() {
     public void setStatisticsItemStateGetter(StatisticsItemStateGetter statisticsItemStateGetter) {
         this.statisticsItemStateGetter = statisticsItemStateGetter;
     }
+
+    public void shutdown() {
+        executor.shutdown();
+    }
 }