[ISSUE #9875] Optimize the RocksDB config shutdown logic when useSingleRocksDBForAllConfigs is set to true to prevent JVM crashes. (#9874)

* Optimize the RocksDB config shutdown logic when useSingleRocksDBForAllConfigs is set to true to prevent JVM crashes.

Change-Id: I309e8d13b6adc46d68146c05ffd7e026e2852ad8

* Fix bug

Change-Id: Ie577e32f65a3902dd60d654f80a8e7eda5790fbf

---------

Co-authored-by: RongtongJin <[email protected]>

Author: RongtongJin

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConfigManager.java

@@ -38,7 +38,6 @@ public class RocksDBConfigManager {
 
     public static final Charset CHARSET = StandardCharsets.UTF_8;
 
-    public volatile boolean isStop = false;
     public ConfigRocksDBStorage configRocksDBStorage = null;
     private FlushOptions flushOptions = null;
     private volatile long lastFlushMemTableMicroSecond = 0;
@@ -72,11 +71,14 @@ public RocksDBConfigManager(String filePath, long memTableFlushInterval, Compres
     }
 
     public boolean init(boolean readOnly) {
-        this.isStop = false;
         this.configRocksDBStorage = ConfigRocksDBStorage.getStore(filePath, readOnly, compressionType);
         return this.configRocksDBStorage.start();
     }
 
+    public boolean isLoaded() {
+        return this.configRocksDBStorage != null && this.configRocksDBStorage.isLoaded();
+    }
+
     public boolean init() {
         return this.init(false);
     }
@@ -113,7 +115,6 @@ public void start() {
     }
 
     public boolean stop() {
-        this.isStop = true;
         ConfigRocksDBStorage.shutdown(filePath);
         if (this.flushOptions != null) {
             this.flushOptions.close();
@@ -123,7 +124,7 @@ public boolean stop() {
 
     public void flushWAL() {
         try {
-            if (this.isStop) {
+            if (!isLoaded()) {
                 return;
             }
             if (this.configRocksDBStorage != null) {
@@ -183,4 +184,5 @@ public Statistics getStatistics() {
 
         return configRocksDBStorage.getStatistics();
     }
+
 }

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConsumerOffsetManager.java

@@ -157,7 +157,7 @@ public String configFilePath() {
 
     @Override
     public synchronized void persist() {
-        if (!rocksDBConfigManager.isStop) {
+        if (rocksDBConfigManager.isLoaded()) {
             try (WriteBatch writeBatch = new WriteBatch()) {
                 for (Entry<String, ConcurrentMap<Integer, Long>> entry : this.offsetTable.entrySet()) {
                     putWriteBatch(writeBatch, entry.getKey(), entry.getValue());

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

@@ -481,6 +481,10 @@ public synchronized boolean start() {
      */
     protected abstract void preShutdown();
 
+    public boolean isLoaded() {
+        return loaded;
+    }
+
     public synchronized boolean shutdown() {
         try {
             if (!this.loaded) {