feat: support shutdown broker gracefully

broker/src/main/java/com/automq/rocketmq/broker/BrokerStartup.java

@@ -18,15 +18,19 @@
 package com.automq.rocketmq.broker;
 
 import com.automq.rocketmq.common.config.BrokerConfig;
+import com.google.common.base.Stopwatch;
 import com.google.common.base.Strings;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.DefaultParser;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
+import org.apache.logging.log4j.LogManager;
 import org.apache.rocketmq.logging.ch.qos.logback.classic.ClassicConstants;
 import org.apache.rocketmq.srvutil.ServerUtil;
 import org.slf4j.Logger;
@@ -77,15 +81,23 @@ public static void main(String[] args) throws Exception {
 
         brokerConfig.validate();
 
-        start(buildBrokerController(brokerConfig));
+        BrokerController controller = buildBrokerController(brokerConfig);
+
+        // Register shutdown hook
+        Runtime.getRuntime().addShutdownHook(new Thread(buildShutdownHook(controller), "ShutdownHook"));
+
+        // Start the broker
+        start(controller);
+
         LOGGER.info("Broker started, costs {} ms", System.currentTimeMillis() - start);
     }
 
-    private static void start(BrokerController controller) {
+    private static void start(BrokerController controller) throws Exception {
         try {
             controller.start();
         } catch (Exception e) {
-            LOGGER.error("Failed to start broker", e);
+            LOGGER.error("Failed to start broker, try to shutdown", e);
+            controller.shutdown();
             System.exit(-1);
         }
     }
@@ -115,6 +127,31 @@ private static void configKernelLogger() {
         }
     }
 
+    public static Runnable buildShutdownHook(BrokerController brokerController) {
+        return new Runnable() {
+            private volatile boolean hasShutdown = false;
+            private final AtomicInteger shutdownTimes = new AtomicInteger(0);
+
+            @Override
+            public void run() {
+                synchronized (this) {
+                    LOGGER.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
+                    if (!this.hasShutdown) {
+                        this.hasShutdown = true;
+                        Stopwatch stopwatch = Stopwatch.createStarted();
+                        try {
+                            brokerController.shutdown();
+                        } catch (Exception e) {
+                            LOGGER.error("Shutdown exception", e);
+                        }
+                        LOGGER.info("Shutdown hook over, consuming total time(ms): {}", stopwatch.elapsed(TimeUnit.MILLISECONDS));
+                        LogManager.shutdown();
+                    }
+                }
+            }
+        };
+    }
+
     private static Options buildCommandlineOptions() {
         Options options = new Options();
         Option opt = new Option("h", "help", false, "Print help");

broker/src/main/java/com/automq/rocketmq/broker/MetricsExporter.java

@@ -61,7 +61,7 @@
 
 public class MetricsExporter implements Lifecycle {
     private static final Logger LOGGER = LoggerFactory.getLogger(MetricsExporter.class);
-
+    private volatile boolean started = false;
     private final BrokerConfig brokerConfig;
     private final MetricsConfig metricsConfig;
     private final static Map<String, String> LABEL_MAP = new HashMap<>();
@@ -213,6 +213,8 @@ public void start() {
             .getMeter("rocketmq-meter");
 
         initMetrics();
+
+        this.started = true;
     }
 
     private void registerMetricsView(SdkMeterProviderBuilder providerBuilder) {
@@ -234,6 +236,9 @@ private void initMetrics() {
 
     @Override
     public void shutdown() {
+        if (!started) {
+            return;
+        }
         MetricsExporterType exporterType = MetricsExporterType.valueOf(metricsConfig.exporterType());
         if (exporterType == MetricsExporterType.OTLP_GRPC) {
             periodicMetricReader.forceFlush();

broker/src/main/resources/log4j2.xml

@@ -16,7 +16,7 @@
   ~ limitations under the License.
   -->
 
-<Configuration>
+<Configuration shutdownHook="disable">
     <Properties>
         <Property name="LOG_DIR">${sys:user.home}${sys:file.separator}logs${sys:file.separator}rocketmqlogs</Property>
     </Properties>

store/src/main/java/com/automq/rocketmq/store/S3StreamStore.java

@@ -168,9 +168,9 @@ public void start() throws Exception {
 
     @Override
     public void shutdown() throws Exception {
-        this.storage.shutdown();
-        this.compactionManager.shutdown();
         this.streamClient.shutdown();
+        this.compactionManager.shutdown();
+        this.storage.shutdown();
     }
 
     @Override