[ISSUE #9701] Synchronize metrics shutdown to prevent JVM crashes during broker shutdown (#9702)

* fix: synchronize metrics shutdown to prevent JVM crash

- Change async shutdown to sync blocking wait in BrokerMetricsManager
- Ensure proper shutdown order to avoid race conditions
- Prevent accessing dependencies after they are shutdown
- Use join() with timeout to wait for CompletableFuture completion
- Apply fix to all metrics exporter types (OTLP_GRPC, PROM, LOG)

* fix codestyle

---------

Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>

Author: guyinyou

broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java

@@ -434,20 +434,20 @@ private void registerMetricsView(SdkMeterProviderBuilder providerBuilder) {
         );
 
         List<Double> commitLatencyBuckets = Arrays.asList(
-                1d * 1 * 1 * 5, //5s
-                1d * 1 * 1 * 60, //1min
-                1d * 1 * 10 * 60, //10min
-                1d * 1 * 60 * 60, //1h
-                1d * 12 * 60 * 60, //12h
-                1d * 24 * 60 * 60 //24h
+            1d * 1 * 1 * 5, //5s
+            1d * 1 * 1 * 60, //1min
+            1d * 1 * 10 * 60, //10min
+            1d * 1 * 60 * 60, //1h
+            1d * 12 * 60 * 60, //12h
+            1d * 24 * 60 * 60 //24h
         );
 
         List<Double> createTimeBuckets = Arrays.asList(
-                (double) Duration.ofMillis(10).toMillis(), //10ms
-                (double) Duration.ofMillis(100).toMillis(), //100ms
-                (double) Duration.ofSeconds(1).toMillis(), //1s
-                (double) Duration.ofSeconds(3).toMillis(), //3s
-                (double) Duration.ofSeconds(5).toMillis() //5s
+            (double) Duration.ofMillis(10).toMillis(), //10ms
+            (double) Duration.ofMillis(100).toMillis(), //100ms
+            (double) Duration.ofSeconds(1).toMillis(), //1s
+            (double) Duration.ofSeconds(3).toMillis(), //3s
+            (double) Duration.ofSeconds(5).toMillis() //5s
         );
         InstrumentSelector messageSizeSelector = InstrumentSelector.builder()
             .setType(InstrumentType.HISTOGRAM)
@@ -470,17 +470,17 @@ private void registerMetricsView(SdkMeterProviderBuilder providerBuilder) {
         providerBuilder.registerView(commitLatencySelector, commitLatencyViewBuilder.build());
 
         InstrumentSelector createTopicTimeSelector = InstrumentSelector.builder()
-                .setType(InstrumentType.HISTOGRAM)
-                .setName(HISTOGRAM_TOPIC_CREATE_EXECUTE_TIME)
-                .build();
+            .setType(InstrumentType.HISTOGRAM)
+            .setName(HISTOGRAM_TOPIC_CREATE_EXECUTE_TIME)
+            .build();
         InstrumentSelector createSubGroupTimeSelector = InstrumentSelector.builder()
-                .setType(InstrumentType.HISTOGRAM)
-                .setName(HISTOGRAM_CONSUMER_GROUP_CREATE_EXECUTE_TIME)
-                .build();
+            .setType(InstrumentType.HISTOGRAM)
+            .setName(HISTOGRAM_CONSUMER_GROUP_CREATE_EXECUTE_TIME)
+            .build();
         ViewBuilder createTopicTimeViewBuilder = View.builder()
-                .setAggregation(Aggregation.explicitBucketHistogram(createTimeBuckets));
+            .setAggregation(Aggregation.explicitBucketHistogram(createTimeBuckets));
         ViewBuilder createSubGroupTimeViewBuilder = View.builder()
-                .setAggregation(Aggregation.explicitBucketHistogram(createTimeBuckets));
+            .setAggregation(Aggregation.explicitBucketHistogram(createTimeBuckets));
         // To config the cardinalityLimit for openTelemetry metrics exporting.
         SdkMeterProviderUtil.setCardinalityLimit(createTopicTimeViewBuilder, brokerConfig.getMetricsOtelCardinalityLimit());
         providerBuilder.registerView(createTopicTimeSelector, createTopicTimeViewBuilder.build());
@@ -588,16 +588,16 @@ private void initRequestMetrics() {
             .build();
 
         topicCreateExecuteTime = brokerMeter.histogramBuilder(HISTOGRAM_TOPIC_CREATE_EXECUTE_TIME)
-                .setDescription("The distribution of create topic time")
-                .ofLongs()
-                .setUnit("milliseconds")
-                .build();
+            .setDescription("The distribution of create topic time")
+            .ofLongs()
+            .setUnit("milliseconds")
+            .build();
 
         consumerGroupCreateExecuteTime = brokerMeter.histogramBuilder(HISTOGRAM_CONSUMER_GROUP_CREATE_EXECUTE_TIME)
-                .setDescription("The distribution of create subscription time")
-                .ofLongs()
-                .setUnit("milliseconds")
-                .build();
+            .setDescription("The distribution of create subscription time")
+            .ofLongs()
+            .setUnit("milliseconds")
+            .build();
     }
 
     private void initConnectionMetrics() {
@@ -720,32 +720,33 @@ private void initTransactionMetrics() {
         }
 
         commitMessagesTotal = brokerMeter.counterBuilder(COUNTER_COMMIT_MESSAGES_TOTAL)
-                .setDescription("Total number of commit messages")
-                .build();
+            .setDescription("Total number of commit messages")
+            .build();
 
         rollBackMessagesTotal = brokerMeter.counterBuilder(COUNTER_ROLLBACK_MESSAGES_TOTAL)
-                .setDescription("Total number of rollback messages")
-                .build();
+            .setDescription("Total number of rollback messages")
+            .build();
 
         transactionFinishLatency = brokerMeter.histogramBuilder(HISTOGRAM_FINISH_MSG_LATENCY)
-                .setDescription("Transaction finish latency")
-                .ofLongs()
-                .setUnit("ms")
-                .build();
+            .setDescription("Transaction finish latency")
+            .ofLongs()
+            .setUnit("ms")
+            .build();
 
         halfMessages = brokerMeter.gaugeBuilder(GAUGE_HALF_MESSAGES)
-                .setDescription("Half messages of all topics")
-                .ofLongs()
-                .buildWithCallback(measurement -> {
-                    brokerController.getTransactionalMessageService().getTransactionMetrics().getTransactionCounts()
-                            .forEach((topic, metric) -> {
-                                measurement.record(
-                                        metric.getCount().get(),
-                                        newAttributesBuilder().put(DefaultStoreMetricsConstant.LABEL_TOPIC, topic).build()
-                                );
-                            });
-                });
+            .setDescription("Half messages of all topics")
+            .ofLongs()
+            .buildWithCallback(measurement -> {
+                brokerController.getTransactionalMessageService().getTransactionMetrics().getTransactionCounts()
+                    .forEach((topic, metric) -> {
+                        measurement.record(
+                            metric.getCount().get(),
+                            newAttributesBuilder().put(DefaultStoreMetricsConstant.LABEL_TOPIC, topic).build()
+                        );
+                    });
+            });
     }
+
     private void initOtherMetrics() {
         if (brokerConfig.isEnableRemotingMetrics()) {
             RemotingMetricsManager.initMetrics(brokerMeter, this::newAttributesBuilder);
@@ -759,19 +760,45 @@ private void initOtherMetrics() {
     }
 
     public void shutdown() {
-        if (brokerConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {
-            periodicMetricReader.forceFlush();
-            periodicMetricReader.shutdown();
-            metricExporter.shutdown();
-        }
-        if (brokerConfig.getMetricsExporterType() == MetricsExporterType.PROM) {
-            prometheusHttpServer.forceFlush();
-            prometheusHttpServer.shutdown();
-        }
-        if (brokerConfig.getMetricsExporterType() == MetricsExporterType.LOG) {
-            periodicMetricReader.forceFlush();
-            periodicMetricReader.shutdown();
-            loggingMetricExporter.shutdown();
+        if (brokerConfig.isInBrokerContainer()) {
+            // only rto need
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {
+                while (!periodicMetricReader.forceFlush().join(60, TimeUnit.SECONDS).isDone()) {
+                }
+                while (!periodicMetricReader.shutdown().join(60, TimeUnit.SECONDS).isSuccess()) {
+                }
+                while (!metricExporter.shutdown().join(60, TimeUnit.SECONDS).isSuccess()) {
+                }
+            }
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.PROM) {
+                while (!prometheusHttpServer.forceFlush().join(60, TimeUnit.SECONDS).isDone()) {
+                }
+                while (!prometheusHttpServer.shutdown().join(60, TimeUnit.SECONDS).isSuccess()) {
+                }
+            }
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.LOG) {
+                while (!periodicMetricReader.forceFlush().join(60, TimeUnit.SECONDS).isDone()) {
+                }
+                while (!periodicMetricReader.shutdown().join(60, TimeUnit.SECONDS).isSuccess()) {
+                }
+                while (!loggingMetricExporter.shutdown().join(60, TimeUnit.SECONDS).isSuccess()) {
+                }
+            }
+        } else {
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {
+                periodicMetricReader.forceFlush();
+                periodicMetricReader.shutdown();
+                metricExporter.shutdown();
+            }
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.PROM) {
+                prometheusHttpServer.forceFlush();
+                prometheusHttpServer.shutdown();
+            }
+            if (brokerConfig.getMetricsExporterType() == MetricsExporterType.LOG) {
+                periodicMetricReader.forceFlush();
+                periodicMetricReader.shutdown();
+                loggingMetricExporter.shutdown();
+            }
         }
     }