From deae8c8ef5e25268219977cf242e4da26fde239e Mon Sep 17 00:00:00 2001 From: majialong Date: Mon, 10 Aug 2026 23:04:21 +0800 Subject: [PATCH] [ISSUE #10899] Fix missing message-out metrics for KV POP --- .../broker/pop/PopConsumerService.java | 30 +++++ .../broker/pop/PopConsumerServiceTest.java | 117 +++++++++++++++++- 2 files changed, 146 insertions(+), 1 deletion(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index f72e2ba26f2..184519e0efd 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -19,6 +19,7 @@ import com.alibaba.fastjson2.JSON; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Stopwatch; +import io.opentelemetry.api.common.Attributes; import java.nio.ByteBuffer; import java.nio.file.Paths; import java.util.ArrayList; @@ -55,6 +56,7 @@ import org.apache.rocketmq.common.message.MessageDecoder; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageExtBrokerInner; +import org.apache.rocketmq.common.topic.TopicValidator; import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils; import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil; import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; @@ -69,6 +71,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CONSUMER_GROUP; +import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_RETRY; +import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_IS_SYSTEM; +import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_TOPIC; + public class PopConsumerService extends ServiceThread { private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME); @@ -177,6 +184,10 @@ public PopConsumerContext handleGetMessageResult(PopConsumerContext context, Get } // build response header here context.addGetMessageResult(result, topicId, queueId, retryType, offset); + String requestTopic = retryType == PopConsumerRecord.RetryType.NORMAL_TOPIC ? + topicId : KeyBuilder.parseNormalTopic(topicId, context.getGroupId()); + this.recordPopMessageOut(result, requestTopic, topicId, context.getGroupId(), + retryType != PopConsumerRecord.RetryType.NORMAL_TOPIC); if (brokerConfig.isPopConsumerKVServiceLog()) { log.info("PopConsumerService pop, time={}, invisible={}, " + "groupId={}, topic={}, queueId={}, offset={}, attemptId={}", @@ -205,6 +216,25 @@ public PopConsumerContext handleGetMessageResult(PopConsumerContext context, Get return context; } + private void recordPopMessageOut(GetMessageResult getMessageResult, String requestTopic, String storeTopic, + String groupId, boolean isRetry) { + int messageCount = getMessageResult.getMessageCount(); + int messageSize = getMessageResult.getBufferTotalSize(); + this.brokerController.getBrokerStatsManager().incBrokerGetNums(requestTopic, messageCount); + this.brokerController.getBrokerStatsManager().incGroupGetNums(groupId, storeTopic, messageCount); + this.brokerController.getBrokerStatsManager().incGroupGetSize(groupId, storeTopic, messageSize); + + Attributes attributes = this.brokerController.getBrokerMetricsManager().newAttributesBuilder() + .put(LABEL_TOPIC, requestTopic) + .put(LABEL_CONSUMER_GROUP, groupId) + .put(LABEL_IS_SYSTEM, + TopicValidator.isSystemTopic(requestTopic) || MixAll.isSysConsumerGroup(groupId)) + .put(LABEL_IS_RETRY, isRetry) + .build(); + this.brokerController.getBrokerMetricsManager().getMessagesOutTotal().add(messageCount, attributes); + this.brokerController.getBrokerMetricsManager().getThroughputOutTotal().add(messageSize, attributes); + } + public long getPopOffset(String groupId, String topicId, int queueId, int initMode, boolean fifo) { // For FIFO messages, the pull offset is not used. diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java index 44189744b46..036c64b8b85 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java @@ -16,6 +16,8 @@ */ package org.apache.rocketmq.broker.pop; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; @@ -34,6 +36,7 @@ import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.failover.EscapeBridge; import org.apache.rocketmq.broker.longpolling.PopLongPollingService; +import org.apache.rocketmq.broker.metrics.BrokerMetricsManager; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; import org.apache.rocketmq.broker.processor.PopMessageProcessor; @@ -41,6 +44,7 @@ import org.apache.rocketmq.broker.topic.TopicConfigManager; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.KeyBuilder; +import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.common.constant.ConsumeInitMode; import org.apache.rocketmq.common.constant.PermName; @@ -65,6 +69,10 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mockito; +import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CONSUMER_GROUP_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_TOPIC_KEY; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; @@ -80,6 +88,10 @@ public class PopConsumerServiceTest { private final String filePath = PopConsumerRocksdbStoreTest.getRandomStorePath(); private BrokerController brokerController; + private BrokerStatsManager brokerStatsManager; + private BrokerMetricsManager brokerMetricsManager; + private LongCounter messagesOutTotal; + private LongCounter throughputOutTotal; private PopConsumerService consumerService; @Before @@ -100,6 +112,10 @@ public void init() throws IOException { PopMessageProcessor popMessageProcessor = Mockito.mock(PopMessageProcessor.class); PopLongPollingService popLongPollingService = Mockito.mock(PopLongPollingService.class); ConsumerOrderInfoManager consumerOrderInfoManager = Mockito.mock(ConsumerOrderInfoManager.class); + brokerStatsManager = Mockito.mock(BrokerStatsManager.class); + brokerMetricsManager = Mockito.mock(BrokerMetricsManager.class); + messagesOutTotal = Mockito.mock(LongCounter.class); + throughputOutTotal = Mockito.mock(LongCounter.class); brokerController = Mockito.mock(BrokerController.class); Mockito.when(brokerController.getBrokerConfig()).thenReturn(brokerConfig); @@ -110,6 +126,11 @@ public void init() throws IOException { Mockito.when(brokerController.getPopMessageProcessor()).thenReturn(popMessageProcessor); Mockito.when(popMessageProcessor.getPopLongPollingService()).thenReturn(popLongPollingService); Mockito.when(brokerController.getConsumerOrderInfoManager()).thenReturn(consumerOrderInfoManager); + Mockito.when(brokerController.getBrokerStatsManager()).thenReturn(brokerStatsManager); + Mockito.when(brokerController.getBrokerMetricsManager()).thenReturn(brokerMetricsManager); + Mockito.when(brokerMetricsManager.newAttributesBuilder()).thenAnswer(invocation -> Attributes.builder()); + Mockito.when(brokerMetricsManager.getMessagesOutTotal()).thenReturn(messagesOutTotal); + Mockito.when(brokerMetricsManager.getThroughputOutTotal()).thenReturn(throughputOutTotal); consumerService = new PopConsumerService(brokerController); } @@ -204,6 +225,100 @@ public void addGetMessageResultTest() { Assert.assertEquals(1, context.getGetMessageResultList().size()); } + @Test + public void handleGetMessageResultShouldRecordMessageOutMetrics() { + PopConsumerContext context = new PopConsumerContext(clientHost, System.currentTimeMillis(), 20000, groupId, + false, ConsumeInitMode.MIN, attemptId); + GetMessageResult result = createFoundGetMessageResult(16, 2, 100L); + consumerService.handleGetMessageResult(context, result, topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100); + + String retryTopic = KeyBuilder.buildPopRetryTopicV2(topicId, groupId); + GetMessageResult retryResult = createFoundGetMessageResult(8, 1, 102L); + consumerService.handleGetMessageResult(context, retryResult, retryTopic, queueId, PopConsumerRecord.RetryType.RETRY_TOPIC_V2, 102); + + Mockito.verify(brokerStatsManager).incBrokerGetNums(topicId, 2); + Mockito.verify(brokerStatsManager).incBrokerGetNums(topicId, 1); + Mockito.verify(brokerStatsManager).incGroupGetNums(groupId, topicId, 2); + Mockito.verify(brokerStatsManager).incGroupGetNums(groupId, retryTopic, 1); + Mockito.verify(brokerStatsManager).incGroupGetSize(groupId, topicId, 16); + Mockito.verify(brokerStatsManager).incGroupGetSize(groupId, retryTopic, 8); + Mockito.verify(throughputOutTotal).add(Mockito.eq(16L), any(Attributes.class)); + Mockito.verify(throughputOutTotal).add(Mockito.eq(8L), any(Attributes.class)); + + ArgumentCaptor attributesCaptor = ArgumentCaptor.forClass(Attributes.class); + Mockito.verify(messagesOutTotal).add(Mockito.eq(2L), attributesCaptor.capture()); + Mockito.verify(messagesOutTotal).add(Mockito.eq(1L), attributesCaptor.capture()); + Assert.assertEquals(topicId, attributesCaptor.getAllValues().get(0).get(LABEL_TOPIC_KEY)); + Assert.assertEquals(groupId, attributesCaptor.getAllValues().get(0).get(LABEL_CONSUMER_GROUP_KEY)); + Assert.assertFalse(attributesCaptor.getAllValues().get(0).get(LABEL_IS_RETRY_KEY)); + Assert.assertFalse(attributesCaptor.getAllValues().get(0).get(LABEL_IS_SYSTEM_KEY)); + + Attributes retryAttributes = attributesCaptor.getAllValues().get(1); + Assert.assertEquals(topicId, retryAttributes.get(LABEL_TOPIC_KEY)); + Assert.assertEquals(groupId, retryAttributes.get(LABEL_CONSUMER_GROUP_KEY)); + Assert.assertFalse(retryAttributes.get(LABEL_IS_SYSTEM_KEY)); + Assert.assertTrue(retryAttributes.get(LABEL_IS_RETRY_KEY)); + } + + @Test + public void handleGetMessageResultShouldRecordRetryV1Topics() { + String retryTopic = KeyBuilder.buildPopRetryTopicV1(topicId, groupId); + PopConsumerContext retryContext = new PopConsumerContext(clientHost, System.currentTimeMillis(), 20000, groupId, + false, ConsumeInitMode.MIN, attemptId); + consumerService.handleGetMessageResult(retryContext, createFoundGetMessageResult(8, 1, 100L), + retryTopic, queueId, PopConsumerRecord.RetryType.RETRY_TOPIC_V1, 100); + + Mockito.verify(brokerStatsManager).incBrokerGetNums(topicId, 1); + Mockito.verify(brokerStatsManager).incGroupGetNums(groupId, retryTopic, 1); + ArgumentCaptor attributesCaptor = ArgumentCaptor.forClass(Attributes.class); + Mockito.verify(messagesOutTotal).add(Mockito.eq(1L), attributesCaptor.capture()); + Assert.assertEquals(topicId, attributesCaptor.getValue().get(LABEL_TOPIC_KEY)); + Assert.assertTrue(attributesCaptor.getValue().get(LABEL_IS_RETRY_KEY)); + } + + @Test + public void handleGetMessageResultShouldNotParseNormalTopicWithRetryPrefix() { + String normalTopic = MixAll.RETRY_GROUP_TOPIC_PREFIX + "normalTopic"; + PopConsumerContext normalContext = new PopConsumerContext(clientHost, System.currentTimeMillis(), 20000, groupId, + false, ConsumeInitMode.MIN, attemptId); + consumerService.handleGetMessageResult(normalContext, createFoundGetMessageResult(8, 1, 101L), + normalTopic, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 101); + + Mockito.verify(brokerStatsManager).incBrokerGetNums(normalTopic, 1); + Mockito.verify(brokerStatsManager).incGroupGetNums(groupId, normalTopic, 1); + ArgumentCaptor attributesCaptor = ArgumentCaptor.forClass(Attributes.class); + Mockito.verify(messagesOutTotal).add(Mockito.eq(1L), attributesCaptor.capture()); + Assert.assertEquals(normalTopic, attributesCaptor.getValue().get(LABEL_TOPIC_KEY)); + Assert.assertFalse(attributesCaptor.getValue().get(LABEL_IS_RETRY_KEY)); + } + + @Test + public void handleGetMessageResultShouldNotRecordEmptyOrNotFoundResult() { + PopConsumerContext context = new PopConsumerContext(clientHost, System.currentTimeMillis(), 20000, groupId, + false, ConsumeInitMode.MIN, attemptId); + GetMessageResult notFoundResult = new GetMessageResult(); + notFoundResult.setStatus(GetMessageStatus.NO_MESSAGE_IN_QUEUE); + consumerService.handleGetMessageResult(context, notFoundResult, topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100); + + GetMessageResult emptyFoundResult = new GetMessageResult(); + emptyFoundResult.setStatus(GetMessageStatus.FOUND); + consumerService.handleGetMessageResult(context, emptyFoundResult, topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100); + + Mockito.verify(brokerStatsManager, Mockito.never()).incBrokerGetNums(anyString(), anyInt()); + Mockito.verify(brokerStatsManager, Mockito.never()).incGroupGetNums(anyString(), anyString(), anyInt()); + Mockito.verify(brokerStatsManager, Mockito.never()).incGroupGetSize(anyString(), anyString(), anyInt()); + Mockito.verify(messagesOutTotal, Mockito.never()).add(anyLong(), any(Attributes.class)); + Mockito.verify(throughputOutTotal, Mockito.never()).add(anyLong(), any(Attributes.class)); + } + + private GetMessageResult createFoundGetMessageResult(int size, int batchNum, long queueOffset) { + GetMessageResult result = new GetMessageResult(); + result.setStatus(GetMessageStatus.FOUND); + ByteBuffer buffer = ByteBuffer.allocate(size); + result.addMessage(new SelectMappedBufferResult(0, buffer, buffer.remaining(), null), queueOffset, batchNum); + return result; + } + @Test public void getMessageAsyncTest() throws Exception { MessageStore messageStore = Mockito.mock(MessageStore.class); @@ -743,4 +858,4 @@ public void testReviveRetryWithSuspendFalseMultipleTimes() { messageExt.setReconsumeTimes(capturedMessage.getReconsumeTimes()); } } -} \ No newline at end of file +}