[ISSUE #9699] Optimize shutdown process and resource management

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();
+    }
 }

container/src/main/java/org/apache/rocketmq/container/InnerSalveBrokerController.java

@@ -19,8 +19,6 @@
 
 import com.google.common.base.Preconditions;
 
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.rocketmq.auth.config.AuthConfig;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -29,8 +27,6 @@
 
 public class InnerSalveBrokerController extends InnerBrokerController {
 
-    private final Lock lock = new ReentrantLock();
-
     public InnerSalveBrokerController(final BrokerContainer brokerContainer,
         final BrokerConfig brokerConfig,
         final MessageStoreConfig storeConfig,

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -176,6 +176,10 @@ public boolean load() {
         return result;
     }
 
+    public void cleanResourceAll() {
+        mappedFileQueue.cleanResourcesAll();
+    }
+
     public void start() {
         this.flushManager.start();
         log.info("start commitLog successfully. storeRoot: {}", this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
@@ -187,12 +191,17 @@ public void start() {
     }
 
     public void shutdown() {
-        this.flushManager.shutdown();
-        log.info("shutdown commitLog successfully. storeRoot: {}", this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
-        flushDiskWatcher.shutdown(true);
+        if (this.flushManager != null) {
+            this.flushManager.shutdown();
+        }
+        if (flushDiskWatcher != null) {
+            flushDiskWatcher.shutdown(true);
+        }
         if (this.coldDataCheckService != null) {
             this.coldDataCheckService.shutdown();
         }
+        putMessageThreadLocal.remove();
+        log.info("shutdown commitLog successfully. storeRoot: {}", this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
     }
 
     public long flush() {
@@ -1347,6 +1356,9 @@ private CompletableFuture<PutMessageStatus> handleHA(AppendMessageResult result,
      * According to receive certain message or offset storage time if an error occurs, it returns -1
      */
     public long pickupStoreTimestamp(final long offset, final int size) {
+        if (defaultMessageStore.isShutdown()) {
+            throw new RuntimeException("message store has shutdown");
+        }
         if (offset >= this.getMinOffset() && offset + size <= this.getMaxOffset()) {
             SelectMappedBufferResult result = this.getMessage(offset, size);
             if (null != result) {

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -214,6 +214,7 @@ public class DefaultMessageStore implements MessageStore {
     public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
         final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig,
         final ConcurrentMap<String, TopicConfig> topicConfigTable) throws IOException {
+        stateMachine = new MessageStoreStateMachine(LOGGER);
         this.messageArrivingListener = messageArrivingListener;
         this.brokerConfig = brokerConfig;
         this.messageStoreConfig = messageStoreConfig;
@@ -256,8 +257,6 @@ public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final Br
         lockFile = new RandomAccessFile(file, "rw");
 
         parseDelayLevel();
-
-        stateMachine = new MessageStoreStateMachine(LOGGER);
     }
 
     public ConsumeQueueStoreInterface createConsumeQueueStore() {
@@ -392,7 +391,7 @@ public void start() throws Exception {
 
         lock = lockFile.getChannel().tryLock(0, 1, false);
         if (lock == null || lock.isShared() || !lock.isValid()) {
-            throw new RuntimeException("Lock failed,MQ already started");
+            throw new RuntimeException("Lock failed, MQ already started, lock status: " + lock);
         }
 
         lockFile.getChannel().write(ByteBuffer.wrap("lock".getBytes(StandardCharsets.UTF_8)));
@@ -476,37 +475,63 @@ private void doRecheckReputOffsetFromCq() throws InterruptedException {
 
     @Override
     public void shutdown() {
-        if (!this.shutdown) {
+        if (!this.stateMachine.getCurrentState().equals(MessageStoreStateMachine.MessageStoreState.SHUTDOWN_OK)) {
             this.shutdown = true;
             this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.SHUTDOWN_BEGIN);
 
-            this.scheduledExecutorService.shutdown();
+            if (this.scheduledExecutorService != null) {
+                this.scheduledExecutorService.shutdown();
+            }
+
             this.scheduledCleanQueueExecutorService.shutdown();
 
             try {
                 this.scheduledExecutorService.awaitTermination(3, TimeUnit.SECONDS);
                 this.scheduledCleanQueueExecutorService.awaitTermination(3, TimeUnit.SECONDS);
                 Thread.sleep(1000 * 3);
-            } catch (InterruptedException e) {
+            } catch (Exception e) {
                 LOGGER.error("shutdown Exception, ", e);
             }
 
             if (this.haService != null) {
                 this.haService.shutdown();
             }
 
-            this.storeStatsService.shutdown();
-            this.commitLog.shutdown();
-            this.reputMessageService.shutdown();
-            this.consumeQueueStore.shutdown();
+            if (this.storeStatsService != null) {
+                this.storeStatsService.shutdown();
+            }
+
+            if (this.commitLog != null) {
+                this.commitLog.shutdown();
+            }
+
+            if (this.reputMessageService != null) {
+                this.reputMessageService.shutdown();
+            }
+
+            if (this.consumeQueueStore != null) {
+                this.consumeQueueStore.shutdown();
+            }
+
             // dispatch-related services must be shut down after reputMessageService
-            this.indexService.shutdown();
+            if (this.indexService != null) {
+                this.indexService.shutdown();
+            }
+
             if (this.compactionService != null) {
                 this.compactionService.shutdown();
             }
-            this.allocateMappedFileService.shutdown();
-            this.storeCheckpoint.shutdown();
+
+            if (this.allocateMappedFileService != null) {
+                this.allocateMappedFileService.shutdown();
+            }
+
+            if (this.storeCheckpoint != null) {
+                this.storeCheckpoint.shutdown();
+            }
+
             this.perfs.shutdown();
+
             if (this.runningFlags.isWriteable() && dispatchBehindBytes() == 0) {
                 this.deleteFile(StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir()));
                 shutDownNormal = true;
@@ -516,13 +541,23 @@ public void shutdown() {
             }
         }
 
-        this.transientStorePool.destroy();
+        if (this.transientStorePool != null) {
+            this.transientStorePool.destroy();
+        }
 
-        if (lockFile != null && lock != null) {
+        if (lock != null) {
             try {
                 lock.release();
+            } catch (IOException e) {
+                LOGGER.error("release file lock error", e);
+            }
+        }
+
+        if (lockFile != null) {
+            try {
                 lockFile.close();
-            } catch (IOException ignored) {
+            } catch (Throwable e) {
+                LOGGER.error("lock file close error", e);
             }
         }
     }
@@ -1805,6 +1840,7 @@ private boolean isTempFileExist() {
         File file = new File(fileName);
         return file.exists();
     }
+
     @Override
     public long getTimingMessageCount(String topic) {
         if (null == timerMessageStore) {

store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java

@@ -799,6 +799,12 @@ public void shutdown(final long intervalForcibly) {
         }
     }
 
+    public void cleanResourcesAll() {
+        for (MappedFile mf : this.mappedFiles) {
+            mf.cleanResources();
+        }
+    }
+
     public void destroy() {
         for (MappedFile mf : this.mappedFiles) {
             mf.destroy(1000 * 3);