[ISSUE apache#10481] Cache OTel Attributes objects in broker/remoting metrics hot path

Builds on top of PR apache#10443 (which introduces the typed AttributeKey constants).

After AttributeKey instances are de-duplicated, the next allocation hot
spot in the metrics path is the Attributes object itself. For repeated
parameter combinations the broker keeps rebuilding identical immutable
Attributes (each ~200-300 byte). This commit introduces a two-level
cache (volatile inline cache + ConcurrentHashMap fallback) at three
points:

- BrokerMetricsManager.getOrBuildTopicAttributes(topic, msgType, isSystem):
  4-field volatile inline cache + CHM keyed by 'topic|msgType|isSystem'.

- RemotingMetricsManager.getOrBuildAttributes(reqCode, respCode, isLongPolling, result):
  long-packed cache key (35 bits across 4 args, no String allocation),
  volatile inline cache + CHM. Falls back to direct build for unknown
  result values.

- RemotingCodeDistributionHandler: replace two independent volatile fields
  (lastCode, lastAdder) with an immutable CachedCounter holder published
  through a single volatile reference. A single volatile read per inc()
  returns a self-consistent (code, adder) snapshot, eliminating a
  torn-read between the two fields under @sharable concurrent access
  from multiple Netty EventLoop threads.

Author: wangjiahua.wjh

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

@@ -72,6 +72,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
@@ -105,6 +106,7 @@
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CONSUME_MODE_KEY;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_RETRY_KEY;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_SYSTEM_KEY;
+import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_MESSAGE_TYPE_KEY;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_LANGUAGE_KEY;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_ID;
 import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_TYPE;
@@ -164,6 +166,12 @@ public class BrokerMetricsManager {
     private LongCounter rollBackMessagesTotal = new NopLongCounter();
     private LongHistogram transactionFinishLatency = new NopLongHistogram();
 
+    private final ConcurrentHashMap<String, Attributes> topicAttributesCache = new ConcurrentHashMap<>();
+    private volatile String lastTopicName;
+    private volatile String lastTopicMsgType;
+    private volatile boolean lastTopicIsSystem;
+    private volatile Attributes lastTopicAttributes;
+
     private final RemotingMetricsManager remotingMetricsManager;
     private final PopMetricsManager popMetricsManager;
 
@@ -195,6 +203,26 @@ public AttributesBuilder newAttributesBuilder() {
         return attributesBuilder;
     }
 
+    public Attributes getOrBuildTopicAttributes(String topic, String messageType, boolean isSystem) {
+        Attributes lastAttrs = this.lastTopicAttributes;
+        if (lastAttrs != null && isSystem == this.lastTopicIsSystem && topic.equals(this.lastTopicName) && messageType.equals(this.lastTopicMsgType)) {
+            return lastAttrs;
+        }
+        String cacheKey = topic + '|' + messageType + '|' + isSystem;
+        Attributes attrs = topicAttributesCache.computeIfAbsent(cacheKey, k ->
+            newAttributesBuilder()
+                .put(LABEL_TOPIC_KEY, topic)
+                .put(LABEL_MESSAGE_TYPE_KEY, messageType)
+                .put(LABEL_IS_SYSTEM_KEY, isSystem)
+                .build()
+        );
+        this.lastTopicName = topic;
+        this.lastTopicMsgType = messageType;
+        this.lastTopicIsSystem = isSystem;
+        this.lastTopicAttributes = attrs;
+        return attrs;
+    }
+
     private Attributes buildLagAttributes(ConsumerLagCalculator.BaseCalculateResult result) {
         AttributesBuilder attributesBuilder = newAttributesBuilder();
         attributesBuilder.put(LABEL_CONSUMER_GROUP_KEY, result.group);

remoting/src/main/java/org/apache/rocketmq/remoting/metrics/RemotingMetricsManager.java

@@ -30,21 +30,30 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.metrics.NopLongHistogram;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
 
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.HISTOGRAM_RPC_LATENCY;
+import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_IS_LONG_POLLING_KEY;
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_PROTOCOL_TYPE_KEY;
+import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_REQUEST_CODE_KEY;
+import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_RESPONSE_CODE_KEY;
+import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_RESULT_KEY;
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.PROTOCOL_TYPE_REMOTING;
-
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.RESULT_CANCELED;
+import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.RESULT_ONEWAY;
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.RESULT_SUCCESS;
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.RESULT_WRITE_CHANNEL_FAILED;
 
 public class RemotingMetricsManager {
     private LongHistogram rpcLatency = new NopLongHistogram();
     private Supplier<AttributesBuilder> attributesBuilderSupplier;
+    private final ConcurrentHashMap<Long, Attributes> attributesCache = new ConcurrentHashMap<>();
+    private volatile long lastAttrsCacheKey;
+    private volatile Attributes lastCachedAttrs;
 
     public RemotingMetricsManager() {
     }
@@ -87,6 +96,44 @@ public List<Pair<InstrumentSelector, ViewBuilder>> getMetricsView() {
         return Lists.newArrayList(new Pair<>(selector, viewBuilder));
     }
 
+    public Attributes getOrBuildAttributes(int requestCode, int responseCode,
+        boolean isLongPolling, String result) {
+        int resultIdx;
+        if (RESULT_SUCCESS.equals(result)) resultIdx = 0;
+        else if (RESULT_ONEWAY.equals(result)) resultIdx = 1;
+        else if (RESULT_WRITE_CHANNEL_FAILED.equals(result)) resultIdx = 2;
+        else if (RESULT_CANCELED.equals(result)) resultIdx = 3;
+        else resultIdx = -1;
+
+        if (resultIdx < 0) {
+            return buildAttributes(requestCode, responseCode, isLongPolling, result);
+        }
+
+        long key = ((long) requestCode << 19)
+            | ((long) (responseCode & 0xFFFF) << 3)
+            | (isLongPolling ? 4L : 0L)
+            | resultIdx;
+        Attributes cached = this.lastCachedAttrs;
+        if (cached != null && key == this.lastAttrsCacheKey) {
+            return cached;
+        }
+        Attributes attrs = attributesCache.computeIfAbsent(key,
+            k -> buildAttributes(requestCode, responseCode, isLongPolling, result));
+        this.lastAttrsCacheKey = key;
+        this.lastCachedAttrs = attrs;
+        return attrs;
+    }
+
+    private Attributes buildAttributes(int requestCode, int responseCode,
+        boolean isLongPolling, String result) {
+        return newAttributesBuilder()
+            .put(LABEL_IS_LONG_POLLING_KEY, isLongPolling)
+            .put(LABEL_REQUEST_CODE_KEY, RemotingHelper.getRequestCodeDesc(requestCode))
+            .put(LABEL_RESPONSE_CODE_KEY, RemotingHelper.getResponseCodeDesc(responseCode))
+            .put(LABEL_RESULT_KEY, result)
+            .build();
+    }
+
     public String getWriteAndFlushResult(Future<?> future) {
         String result = RESULT_SUCCESS;
         if (future.isCancelled()) {

remoting/src/main/java/org/apache/rocketmq/remoting/netty/RemotingCodeDistributionHandler.java

@@ -30,21 +30,50 @@
 @ChannelHandler.Sharable
 public class RemotingCodeDistributionHandler extends ChannelDuplexHandler {
 
+    /**
+     * Immutable holder for a cached code→adder pair.  Published via a single volatile
+     * write so racing threads always observe a consistent pair (never a mixed code from
+     * one update with an adder from another).
+     */
+    private static final class CachedCounter {
+        final int code;
+        final LongAdder adder;
+
+        CachedCounter(int code, LongAdder adder) {
+            this.code = code;
+            this.adder = adder;
+        }
+    }
+
     private final ConcurrentMap<Integer, LongAdder> inboundDistribution;
     private final ConcurrentMap<Integer, LongAdder> outboundDistribution;
+    private volatile CachedCounter lastIn;
+    private volatile CachedCounter lastOut;
 
     public RemotingCodeDistributionHandler() {
         inboundDistribution = new ConcurrentHashMap<>();
         outboundDistribution = new ConcurrentHashMap<>();
     }
 
     private void countInbound(int requestCode) {
+        CachedCounter cached = lastIn;
+        if (cached != null && cached.code == requestCode) {
+            cached.adder.increment();
+            return;
+        }
         LongAdder item = inboundDistribution.computeIfAbsent(requestCode, k -> new LongAdder());
+        lastIn = new CachedCounter(requestCode, item);
         item.increment();
     }
 
     private void countOutbound(int responseCode) {
+        CachedCounter cached = lastOut;
+        if (cached != null && cached.code == responseCode) {
+            cached.adder.increment();
+            return;
+        }
         LongAdder item = outboundDistribution.computeIfAbsent(responseCode, k -> new LongAdder());
+        lastOut = new CachedCounter(responseCode, item);
         item.increment();
     }