[ISSUE #10266] Fix OOM caused by OpenTelemetry 1.44 OtlpGrpcMetricExporter pool race (#10267)

OpenTelemetry Java 1.44.0 ~ 1.46.x ships OtlpGrpcMetricExporter with
MemoryMode.REUSABLE_DATA by default. The underlying
MetricReusableDataMarshaler.marshalerPool is a non-thread-safe
ArrayDeque accessed concurrently by the reader thread (poll) and the
OkHttp callback thread (add, via whenComplete). With
BatchSplittingMetricExporter issuing N concurrent sub-batch exports
per cycle, the pool races and leaks marshalers (~132 KiB each) until
OOM. Fixed upstream in 1.47.0 via open-telemetry/opentelemetry-java#7041
(ArrayDeque -> ConcurrentLinkedDeque).

- Bump OpenTelemetry to 1.47.0 in pom.xml so the upstream race fix is
  in effect.
- Default OtlpGrpcMetricExporter to MemoryMode.IMMUTABLE_DATA to
  preserve the pre-1.44 default behavior; exposed via
  brokerConfig.metricsExportOtelMemoryMode ("IMMUTABLE_DATA" /
  "REUSABLE_DATA", case-insensitive). Operators may opt in to
  REUSABLE_DATA when running on OTel >= 1.47.
- Cap concurrent in-flight sub-batches in BatchSplittingMetricExporter
  with a Semaphore controlled by
  brokerConfig.metricsExportBatchMaxConcurrent (default 4; set to 1
  to serialize and match pre-batch behavior; 0 or Integer.MAX_VALUE
  means unlimited).
- Add brokerConfig.metricsExportBatchSplitEnabled (default true) as
  an escape hatch to bypass BatchSplittingMetricExporter entirely,
  restoring the raw OtlpGrpcMetricExporter wiring.
- Defensively snapshot MetricData points before export to avoid
  ArrayIndexOutOfBoundsException in NumberDataPointMarshaler when
  async instrument callbacks mutate point collections during export.

Author: Houlong66

WORKSPACE

@@ -89,15 +89,15 @@ maven_install(
         "io.grpc:grpc-api:1.47.0",
         "io.grpc:grpc-testing:1.47.0",
         "org.springframework:spring-core:5.3.26",
-        "io.opentelemetry:opentelemetry-exporter-otlp:1.29.0",
-        "io.opentelemetry:opentelemetry-exporter-prometheus:1.29.0-alpha",
-        "io.opentelemetry:opentelemetry-exporter-logging:1.29.0",
-        "io.opentelemetry:opentelemetry-sdk:1.29.0",
-        "io.opentelemetry:opentelemetry-exporter-logging-otlp:1.29.0",
+        "io.opentelemetry:opentelemetry-exporter-otlp:1.47.0",
+        "io.opentelemetry:opentelemetry-exporter-prometheus:1.47.0-alpha",
+        "io.opentelemetry:opentelemetry-exporter-logging:1.47.0",
+        "io.opentelemetry:opentelemetry-sdk:1.47.0",
+        "io.opentelemetry:opentelemetry-exporter-logging-otlp:1.47.0",
         "com.squareup.okio:okio-jvm:3.0.0",
-        "io.opentelemetry:opentelemetry-api:1.29.0",
-        "io.opentelemetry:opentelemetry-sdk-metrics:1.29.0",
-        "io.opentelemetry:opentelemetry-sdk-common:1.29.0",
+        "io.opentelemetry:opentelemetry-api:1.47.0",
+        "io.opentelemetry:opentelemetry-sdk-metrics:1.47.0",
+        "io.opentelemetry:opentelemetry-sdk-common:1.47.0",
         "io.github.aliyunmq:rocketmq-slf4j-api:1.0.0",
         "io.github.aliyunmq:rocketmq-logback-classic:1.0.0",
         "org.slf4j:jul-to-slf4j:2.0.6",

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

@@ -46,6 +46,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.Semaphore;
 import java.util.function.IntSupplier;
 
 /**
@@ -70,6 +71,16 @@
  * multiple smaller MetricData objects, each preserving the
  * original resource, scope, name, description, unit, and
  * type metadata.
+ *
+ * <p>Sub-batch submissions are bounded by a per-call
+ * semaphore whose permit count comes from
+ * {@code maxConcurrentSupplier}. This caps the number of
+ * batches that can be in-flight to the delegate at once,
+ * preventing the MetricData retention window from being
+ * multiplied by the batch count under OTel SDK 1.31+ where
+ * the OTLP exporter may hold references until the RPC
+ * completes. A value &lt;= 0 or Integer.MAX_VALUE disables
+ * the limit (legacy behavior).
  */
 public final class BatchSplittingMetricExporter
     implements MetricExporter {
@@ -85,16 +96,24 @@ public final class BatchSplittingMetricExporter
     /** Supplies the max data points per batch at runtime. */
     private final IntSupplier maxBatchSizeSupplier;
 
+    /** Supplies the max concurrent in-flight batches. */
+    private final IntSupplier maxConcurrentSupplier;
+
     /**
      * Creates a new BatchSplittingMetricExporter.
      *
      * @param metricExporter the underlying MetricExporter
      * @param batchSizeSupplier supplies the max number
      *     of data points per batch; must return &gt; 0
+     * @param concurrencySupplier supplies the max number
+     *     of sub-batches that can be in-flight to the
+     *     delegate at once; &lt;= 0 or Integer.MAX_VALUE
+     *     means unlimited
      */
     public BatchSplittingMetricExporter(
         final MetricExporter metricExporter,
-        final IntSupplier batchSizeSupplier) {
+        final IntSupplier batchSizeSupplier,
+        final IntSupplier concurrencySupplier) {
         if (metricExporter == null) {
             throw new NullPointerException(
                 "metricExporter must not be null");
@@ -103,8 +122,13 @@ public BatchSplittingMetricExporter(
             throw new NullPointerException(
                 "batchSizeSupplier must not be null");
         }
+        if (concurrencySupplier == null) {
+            throw new NullPointerException(
+                "concurrencySupplier must not be null");
+        }
         this.delegate = metricExporter;
         this.maxBatchSizeSupplier = batchSizeSupplier;
+        this.maxConcurrentSupplier = concurrencySupplier;
     }
 
     /** {@inheritDoc} */
@@ -136,23 +160,48 @@ public CompletableResultCode export(
         List<List<MetricData>> batches =
             splitIntoBatches(snapshotted, maxBatchSize);
 
+        int maxConcurrent =
+            maxConcurrentSupplier.getAsInt();
+        final Semaphore semaphore =
+            (maxConcurrent > 0
+                && maxConcurrent < Integer.MAX_VALUE)
+                ? new Semaphore(maxConcurrent)
+                : null;
+
         LOGGER.debug(
             "Splitting metrics export: "
                 + "totalDataPoints={}, "
                 + "maxBatchSize={}, "
-                + "batchCount={}",
+                + "batchCount={}, "
+                + "maxConcurrent={}",
             totalDataPoints, maxBatchSize,
-            batches.size());
+            batches.size(), maxConcurrent);
 
         List<CompletableResultCode> results =
             new ArrayList<>(batches.size());
         for (int i = 0; i < batches.size(); i++) {
             final List<MetricData> batch =
                 batches.get(i);
             final int batchIndex = i;
+            if (semaphore != null) {
+                try {
+                    semaphore.acquire();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    LOGGER.warn(
+                        "Interrupted while waiting "
+                            + "for export slot; "
+                            + "submitted {} of {} batches",
+                        i, batches.size());
+                    break;
+                }
+            }
             CompletableResultCode r =
                 delegate.export(batch);
             r.whenComplete(() -> {
+                if (semaphore != null) {
+                    semaphore.release();
+                }
                 if (!r.isSuccess()) {
                     logFailedBatch(batchIndex, batch);
                 }

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

@@ -28,6 +28,7 @@
 import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder;
 import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.metrics.Aggregation;
 import io.opentelemetry.sdk.metrics.InstrumentSelector;
 import io.opentelemetry.sdk.metrics.InstrumentType;
@@ -69,6 +70,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -318,6 +320,19 @@ private boolean checkConfig() {
         return false;
     }
 
+    private static MemoryMode resolveMemoryMode(String configured) {
+        if (StringUtils.isBlank(configured)) {
+            return MemoryMode.IMMUTABLE_DATA;
+        }
+        try {
+            return MemoryMode.valueOf(configured.trim().toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            LOGGER.warn("Invalid metricsExportOtelMemoryMode '{}', falling back to IMMUTABLE_DATA. Valid values: IMMUTABLE_DATA, REUSABLE_DATA.",
+                configured);
+            return MemoryMode.IMMUTABLE_DATA;
+        }
+    }
+
     private void init() {
         MetricsExporterType metricsExporterType = brokerConfig.getMetricsExporterType();
         if (metricsExporterType == MetricsExporterType.DISABLE) {
@@ -356,9 +371,16 @@ private void init() {
             if (!endpoint.startsWith("http")) {
                 endpoint = "https://" + endpoint;
             }
+            // OTel 1.44.0 ~ 1.46.x defaults OtlpGrpcMetricExporter to REUSABLE_DATA,
+            // whose MetricReusableDataMarshaler uses a non-thread-safe ArrayDeque pool.
+            // Combined with BatchSplittingMetricExporter's concurrent sub-batch export
+            // this triggers a pool race that leaks marshalers until OOM (fixed upstream
+            // in 1.47.0 via opentelemetry-java#7041). IMMUTABLE_DATA bypasses that path.
+            MemoryMode memoryMode = resolveMemoryMode(brokerConfig.getMetricsExportOtelMemoryMode());
             OtlpGrpcMetricExporterBuilder metricExporterBuilder = OtlpGrpcMetricExporter.builder()
                 .setEndpoint(endpoint)
                 .setCompression("gzip")
+                .setMemoryMode(memoryMode)
                 .setTimeout(brokerConfig.getMetricGrpcExporterTimeOutInMills(), TimeUnit.MILLISECONDS)
                 .setAggregationTemporalitySelector(type -> {
                     if (brokerConfig.isMetricsInDelta() &&
@@ -384,8 +406,16 @@ private void init() {
             }
 
             OtlpGrpcMetricExporter otlpExporter = metricExporterBuilder.build();
-            metricExporter = new BatchSplittingMetricExporter(otlpExporter,
-                brokerConfig::getMetricsExportBatchMaxDataPoints);
+            if (brokerConfig.isMetricsExportBatchSplitEnabled()) {
+                metricExporter = new BatchSplittingMetricExporter(otlpExporter,
+                    brokerConfig::getMetricsExportBatchMaxDataPoints,
+                    brokerConfig::getMetricsExportBatchMaxConcurrent);
+            } else {
+                // Escape hatch: skip the splitter wrapper entirely and use the raw
+                // OTLP exporter. Gives up the oversized-payload guard but removes
+                // any splitter-side overhead/risk. Re-enable if needed.
+                metricExporter = otlpExporter;
+            }
 
             periodicMetricReader = PeriodicMetricReader.builder(metricExporter)
                 .setInterval(brokerConfig.getMetricGrpcExporterIntervalInMills(), TimeUnit.MILLISECONDS)