(rocketmq) branch develop updated: [ISSUE #9816] Fix concurrent modify opentelemetry record in calculate consumer lag (#9817)

[lollipop]

This is an automated email from the ASF dual-hosted git repository.

lollipop pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 736d2a860e [ISSUE #9816] Fix concurrent modify opentelemetry record in 
calculate consumer lag (#9817)
736d2a860e is described below

commit 736d2a860e672f1bc6058e13b246c10b80522b12
Author: lizhimins <707364...@qq.com>
AuthorDate: Fri Nov 7 11:47:53 2025 +0800

    [ISSUE #9816] Fix concurrent modify opentelemetry record in calculate 
consumer lag (#9817)
---
 .../org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java  | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git 
a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
 
b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
index a6707d0dd2..960c1dd250 100644
--- 
a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
+++ 
b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
@@ -674,7 +674,13 @@ public class BrokerMetricsManager {
             .setDescription("Consumer lag messages")
             .ofLongs()
             .buildWithCallback(measurement ->
-                consumerLagCalculator.calculateLag(result -> 
measurement.record(result.lag, buildLagAttributes(result))));
+                consumerLagCalculator.calculateLag(result -> {
+                    // Note: 'record' method uses HashMap which may cause
+                    // concurrent access issues when Pull thread executes Pop 
callbacks.
+                    synchronized (this) {
+                        measurement.record(result.lag, 
buildLagAttributes(result));
+                    }
+                }));
 
         consumerLagLatency = 
brokerMeter.gaugeBuilder(GAUGE_CONSUMER_LAG_LATENCY)
             .setDescription("Consumer lag time")