feat(rocketmq): AUT-686 optimize consume stats api (apache#15)

Signed-off-by: SSpirits <admin@lv5.moe>
lyx2000 · Jun 29, 2023 · 04aa146 · 04aa146
1 parent 93250ef
commit 04aa146
Show file tree

Hide file tree

Showing 23 changed files with 91 additions and 21 deletions.
diff --git a/acl/pom.xml b/acl/pom.xml
@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
     <artifactId>rocketmq-acl</artifactId>
     <name>rocketmq-acl ${project.version}</name>

diff --git a/broker/pom.xml b/broker/pom.xml
@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

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
@@ -138,6 +138,7 @@ public class BrokerMetricsManager {
     public static final List<String> SYSTEM_GROUP_PREFIX_LIST = new ArrayList<String>() {
         {
             add(MixAll.CID_RMQ_SYS_PREFIX.toLowerCase());
+            add("automq_housekeeping".toLowerCase());
         }
     };
 

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -46,6 +46,7 @@
 import org.apache.rocketmq.broker.controller.ReplicasManager;
 import org.apache.rocketmq.broker.filter.ConsumerFilterData;
 import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
+import org.apache.rocketmq.broker.metrics.ConsumerLagCalculator;
 import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
@@ -54,6 +55,7 @@
 import org.apache.rocketmq.common.LockCallback;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UnlockCallback;
@@ -161,6 +163,7 @@
 import org.apache.rocketmq.remoting.protocol.header.UpdateGlobalWhiteAddrsConfigRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.UpdateGroupForbiddenRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
 import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.remoting.protocol.statictopic.LogicQueueMappingItem;
 import org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
@@ -192,9 +195,11 @@
 public class AdminBrokerProcessor implements NettyRequestProcessor {
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
     protected final BrokerController brokerController;
+    private final ConsumerLagCalculator consumerLagCalculator;
 
     public AdminBrokerProcessor(final BrokerController brokerController) {
         this.brokerController = brokerController;
+        consumerLagCalculator = new ConsumerLagCalculator(brokerController);
     }
 
     @Override
@@ -1500,6 +1505,14 @@ private RemotingCommand getConsumeStats(ChannelHandlerContext ctx,
             topics.add(requestHeader.getTopic());
         }
 
+        String group = requestHeader.getConsumerGroup();
+        ConsumerGroupInfo consumerGroupInfo = brokerController.getConsumerManager()
+            .getConsumerGroupInfo(group, true);
+        boolean isPop = false;
+        if (consumerGroupInfo != null) {
+            isPop = consumerGroupInfo.getConsumeType() == ConsumeType.CONSUME_POP;
+        }
+
         for (String topic : topics) {
             TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
             if (null == topicConfig) {
@@ -1549,11 +1562,28 @@ private RemotingCommand getConsumeStats(ChannelHandlerContext ctx,
 
                 long pullOffset = this.brokerController.getConsumerOffsetManager().queryPullOffset(
                     requestHeader.getConsumerGroup(), topic, i);
+                if (isPop) {
+                    pullOffset = brokerController.getPopMessageProcessor().getPopBufferMergeService().getLatestOffset(topic, group, i);
+                    if (pullOffset < 0) {
+                        pullOffset = brokerController.getConsumerOffsetManager().queryOffset(group, topic, i);
+                    }
+                    if (pullOffset < 0) {
+                        pullOffset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
+                    }
+                }
 
                 offsetWrapper.setBrokerOffset(brokerOffset);
                 offsetWrapper.setConsumerOffset(consumerOffset);
                 offsetWrapper.setPullOffset(Math.max(consumerOffset, pullOffset));
 
+                Pair<Long, Long> lagStats = consumerLagCalculator.getConsumerLagStats(requestHeader.getConsumerGroup(), topic, i, isPop);
+                offsetWrapper.setLagEstimatedAccumulation(lagStats.getObject1());
+                offsetWrapper.setEarliestUnconsumedTimestamp(lagStats.getObject2());
+
+                Pair<Long, Long> inflightStats = consumerLagCalculator.getInFlightMsgStats(requestHeader.getConsumerGroup(), topic, i, isPop);
+                offsetWrapper.setInFlightMsgCountEstimatedAccumulation(inflightStats.getObject1());
+                offsetWrapper.setEarliestUnPulledTimestamp(inflightStats.getObject2());
+
                 long timeOffset = consumerOffset - 1;
                 if (timeOffset >= 0) {
                     long lastTimestamp = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, timeOffset);

diff --git a/client/pom.xml b/client/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/common/pom.xml b/common/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/container/pom.xml b/container/pom.xml
@@ -18,7 +18,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/controller/pom.xml b/controller/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

diff --git a/distribution/Dockerfile b/distribution/Dockerfile
@@ -3,7 +3,7 @@ FROM openjdk:17-jdk-slim
 RUN apt-get update \
     && apt-get install -y --no-install-recommends curl unzip procps vim inetutils-ping && apt-get clean
 
-ENV ROCKETMQ_VERSION 5.1.1-SNAPSHOT
+ENV ROCKETMQ_VERSION 5.1.1-automq-SNAPSHOT
 ENV ROCKETMQ_HOME  /root/rocketmq/rocketmq-${ROCKETMQ_VERSION}
 
 WORKDIR  ${ROCKETMQ_HOME}

diff --git a/distribution/pom.xml b/distribution/pom.xml
@@ -20,7 +20,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
     <artifactId>rocketmq-distribution</artifactId>
     <name>rocketmq-distribution ${project.version}</name>

diff --git a/example/pom.xml b/example/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/filter/pom.xml b/filter/pom.xml
@@ -20,7 +20,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/namesrv/pom.xml b/namesrv/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/openmessaging/pom.xml b/openmessaging/pom.xml
@@ -20,7 +20,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/pom.xml b/pom.xml
@@ -28,7 +28,7 @@
     <inceptionYear>2012</inceptionYear>
     <groupId>org.apache.rocketmq</groupId>
     <artifactId>rocketmq-all</artifactId>
-    <version>5.1.1</version>
+    <version>5.1.1-automq-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>Apache RocketMQ ${project.version}</name>
     <url>http://rocketmq.apache.org/</url>
@@ -37,7 +37,7 @@
         <url>git@github.com:apache/rocketmq.git</url>
         <connection>scm:git:git@github.com:apache/rocketmq.git</connection>
         <developerConnection>scm:git:git@github.com:apache/rocketmq.git</developerConnection>
-        <tag>rocketmq-all-5.1.1</tag>
+        <tag>HEAD</tag>
     </scm>
 
     <mailingLists>

diff --git a/proxy/pom.xml b/proxy/pom.xml
@@ -20,7 +20,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/remoting/pom.xml b/remoting/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/OffsetWrapper.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/admin/OffsetWrapper.java
@@ -20,7 +20,14 @@ public class OffsetWrapper {
     private long brokerOffset;
     private long consumerOffset;
     private long pullOffset;
+
+
+    private long lagEstimatedAccumulation;
+    private long inFlightMsgCountEstimatedAccumulation;
+
     private long lastTimestamp;
+    private long earliestUnconsumedTimestamp;
+    private long earliestUnPulledTimestamp;
 
     public long getBrokerOffset() {
         return brokerOffset;
@@ -46,11 +53,43 @@ public void setPullOffset(long pullOffset) {
         this.pullOffset = pullOffset;
     }
 
+    public long getLagEstimatedAccumulation() {
+        return lagEstimatedAccumulation;
+    }
+
+    public void setLagEstimatedAccumulation(long lagEstimatedAccumulation) {
+        this.lagEstimatedAccumulation = lagEstimatedAccumulation;
+    }
+
+    public long getInFlightMsgCountEstimatedAccumulation() {
+        return inFlightMsgCountEstimatedAccumulation;
+    }
+
+    public void setInFlightMsgCountEstimatedAccumulation(long inFlightMsgCountEstimatedAccumulation) {
+        this.inFlightMsgCountEstimatedAccumulation = inFlightMsgCountEstimatedAccumulation;
+    }
+
     public long getLastTimestamp() {
         return lastTimestamp;
     }
 
     public void setLastTimestamp(long lastTimestamp) {
         this.lastTimestamp = lastTimestamp;
     }
+
+    public long getEarliestUnconsumedTimestamp() {
+        return earliestUnconsumedTimestamp;
+    }
+
+    public void setEarliestUnconsumedTimestamp(long earliestUnconsumedTimestamp) {
+        this.earliestUnconsumedTimestamp = earliestUnconsumedTimestamp;
+    }
+
+    public long getEarliestUnPulledTimestamp() {
+        return earliestUnPulledTimestamp;
+    }
+
+    public void setEarliestUnPulledTimestamp(long earliestUnPulledTimestamp) {
+        this.earliestUnPulledTimestamp = earliestUnPulledTimestamp;
+    }
 }
diff --git a/srvutil/pom.xml b/srvutil/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/store/pom.xml b/store/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/test/pom.xml b/test/pom.xml
@@ -20,7 +20,7 @@
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/tieredstore/pom.xml b/tieredstore/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/tools/pom.xml b/tools/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>5.1.1</version>
+        <version>5.1.1-automq-SNAPSHOT</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>