[ISSUE #9217] Fix broker's inflight and available message counts incorrect when the pop consumer service is enabled (#9218)

Author: lizhimins

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);
         }