[ISSUE #9838] IndexStoreService use forceShutdown when disk is not writable (#9839)

* Update store and tieredstore implementation

Change-Id: I29c931148110c4238ce9a256ab2bf69d7b6139a8
Co-developed-by: Cursor <noreply@cursor.com>

* Fix UT NPE

Change-Id: I699ec538118166743308af9e78da50f2aa1b56b3

---------

Co-authored-by: RongtongJin <user@example.com>

Author: RongtongJin

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

@@ -85,6 +85,15 @@ public boolean isWriteable() {
         return false;
     }
 
+    public boolean isStoreWriteable() {
+        if ((this.flagBits & NOT_WRITEABLE_BIT) == 0) {
+            return true;
+        }
+
+        return false;
+    }
+
+
     //for consume queue, just ignore the DISK_FULL_BIT
     public boolean isCQWriteable() {
         if ((this.flagBits & (NOT_WRITEABLE_BIT | WRITE_LOGICS_QUEUE_ERROR_BIT | WRITE_INDEX_FILE_ERROR_BIT | LOGIC_DISK_FULL_BIT)) == 0) {
@@ -94,7 +103,7 @@ public boolean isCQWriteable() {
         return false;
     }
 
-    public boolean getAndMakeNotWriteable() {
+    public boolean getAndMakeStoreNotWriteable() {
         boolean result = this.isWriteable();
         if (result) {
             this.flagBits |= NOT_WRITEABLE_BIT;

store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java

@@ -592,7 +592,7 @@ public boolean getAndMakeNotWriteable() {
         if (runningFlags == null) {
             return false;
         }
-        return runningFlags.getAndMakeNotWriteable();
+        return runningFlags.getAndMakeStoreNotWriteable();
     }
 
     public boolean isWriteable() {

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java

@@ -338,7 +338,7 @@ public CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int q
                 .thenApply(time -> {
                     Attributes latencyAttributes = TieredStoreMetricsManager.newAttributesBuilder()
                         .put(TieredStoreMetricsConstant.LABEL_OPERATION,
-                                TieredStoreMetricsConstant.OPERATION_API_GET_TIME_BY_OFFSET)
+                            TieredStoreMetricsConstant.OPERATION_API_GET_TIME_BY_OFFSET)
                         .put(TieredStoreMetricsConstant.LABEL_TOPIC, topic)
                         .build();
                     TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
@@ -465,8 +465,13 @@ public synchronized void shutdown() {
             dispatcher.shutdown();
         }
         if (indexService != null) {
-            indexService.shutdown();
+            if (defaultStore.getRunningFlags() != null && defaultStore.getRunningFlags().isStoreWriteable()) {
+                indexService.shutdown();
+            } else {
+                indexService.forceShutdown();
+            }
         }
+
         if (flatFileStore != null) {
             flatFileStore.shutdown();
         }

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexService.java

@@ -60,6 +60,13 @@ default void forceUpload() {
      */
     void shutdown();
 
+    /**
+     * Force shutdown the index service.
+     */
+    default void forceShutdown() {
+        shutdown();
+    };
+
     /**
      * Destroys the index service and releases all resources.
      */

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java

@@ -414,6 +414,11 @@ public void shutdown() {
         }
     }
 
+    @Override
+    public void forceShutdown() {
+        super.shutdown();
+    }
+
     @Override
     public void run() {
         while (!this.isStopped()) {