[ISSUE apache#9217] Fix broker's inflight and available message count…

…s incorrect when the pop consumer service is enabled (apache#9218)
yx9o · Mar 3, 2025 · 40b25b3 · 40b25b3
1 parent 988c826
commit 40b25b3
Showing 1 changed file with 3 additions and 6 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/metrics/ConsumerLagCalculator.java b/broker/src/main/java/org/apache/rocketmq/broker/metrics/ConsumerLagCalculator.java
@@ -348,7 +348,7 @@ public Pair<Long, Long> getConsumerLagStats(String group, String topic, int queu
             brokerOffset = 0;
         }
 
-        if (isPop) {
+        if (isPop && !brokerConfig.isPopConsumerKVServiceEnable()) {
             long pullOffset = popBufferMergeService.getLatestOffset(topic, group, queueId);
             if (pullOffset < 0) {
                 pullOffset = offsetManager.queryOffset(group, topic, queueId);
@@ -401,7 +401,7 @@ public Pair<Long, Long> getInFlightMsgStats(String group, String topic, boolean
 
     public Pair<Long, Long> getInFlightMsgStats(String group, String topic, int queueId, boolean isPop)
         throws ConsumeQueueException {
-        if (isPop) {
+        if (isPop && !brokerConfig.isPopConsumerKVServiceEnable()) {
             long inflight = popInflightMessageCounter.getGroupPopInFlightMessageNum(topic, group, queueId);
             long pullOffset = popBufferMergeService.getLatestOffset(topic, group, queueId);
             if (pullOffset < 0) {
@@ -456,14 +456,11 @@ public long getAvailableMsgCount(String group, String topic, int queueId, boolea
         }
 
         long pullOffset;
-        if (isPop) {
+        if (isPop && !brokerConfig.isPopConsumerKVServiceEnable()) {
             pullOffset = popBufferMergeService.getLatestOffset(topic, group, queueId);
             if (pullOffset < 0) {
                 pullOffset = offsetManager.queryOffset(group, topic, queueId);
             }
-            if (pullOffset < 0) {
-                pullOffset = brokerOffset;
-            }
         } else {
             pullOffset = offsetManager.queryPullOffset(group, topic, queueId);
         }