Skip to content

[ISSUE #10899] Fix missing message-out metrics for KV POP #10900

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
majialoong wants to merge 2 commits into
base: develop
Choose a base branch
from
Open

[ISSUE #10899] Fix missing message-out metrics for KV POP #10900

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
deae8c8
[ISSUE #10899] Fix missing message-out metrics for KV POP
majialoong Aug 10, 2026
829ce50
Merge branch 'develop' into ISSUE-10899
majialoong Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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);
Expand Down Expand Up @@ -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={}",
Expand Down Expand Up @@ -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.
Expand Down
117 changes: 116 additions & 1 deletion broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -34,13 +36,15 @@
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;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
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;
Expand All @@ -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;
Expand All @@ -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
Expand All @@ -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);
Expand All @@ -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);
}
Expand Down Expand Up @@ -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<Attributes> 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<Attributes> 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<Attributes> 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);
Expand Down Expand Up @@ -743,4 +858,4 @@ public void testReviveRetryWithSuspendFalseMultipleTimes() {
messageExt.setReconsumeTimes(capturedMessage.getReconsumeTimes());
}
}
}
}
Loading