commit 5fb1de22cd12555164103771aa42d93d00768ba5
author dinglei <libya_003@163.com> Wed Nov 09 10:04:31 2022 +0800
committer GitHub <noreply@github.com> Wed Nov 09 10:04:31 2022 +0800
tree b2954d32fc53eda3675a8f645667ff83bf454107
parent 012f3a1dbc741b785f629ce1dfefe8f2e58f7939
parent e0709f53a8df61385abcbd094bc18db987e87102

[ISSUE #99] ignored rocketmq_group_get_latency_by_storetime for boardcast consumer

[ISSUE #100] ignored rocketmq_group_get_latency_by_storetime for boardcast consumer

src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java

diff --git a/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java b/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
index b29c839..7f4b940 100644
--- a/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
+++ b/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
@@ -378,37 +378,39 @@
                 }
 
                 // get consumer latency
-                try {
-                    HashMap<String, Long> consumerLatencyMap = new HashMap<>();
-                    for (Map.Entry<MessageQueue, OffsetWrapper> consumeStatusEntry : consumeStats.getOffsetTable().entrySet()) {
-                        MessageQueue q = consumeStatusEntry.getKey();
-                        OffsetWrapper offset = consumeStatusEntry.getValue();
-                        PullResult consumePullResult = ((MQAdminExtImpl) mqAdminExt).queryMsgByOffset(q, offset.getConsumerOffset());
-                        long lagTime = 0;
-                        if (consumePullResult != null && consumePullResult.getPullStatus() == PullStatus.FOUND) {
-                            lagTime = System.currentTimeMillis() - consumePullResult.getMsgFoundList().get(0).getStoreTimestamp();
-                            if (offset.getBrokerOffset() == offset.getConsumerOffset()) {
-                                lagTime = 0;
+                if (MessageModel.CLUSTERING == messageModel) {
+                    try {
+                        HashMap<String, Long> consumerLatencyMap = new HashMap<>();
+                        for (Map.Entry<MessageQueue, OffsetWrapper> consumeStatusEntry : consumeStats.getOffsetTable().entrySet()) {
+                            MessageQueue q = consumeStatusEntry.getKey();
+                            OffsetWrapper offset = consumeStatusEntry.getValue();
+                            PullResult consumePullResult = ((MQAdminExtImpl) mqAdminExt).queryMsgByOffset(q, offset.getConsumerOffset());
+                            long lagTime = 0;
+                            if (consumePullResult != null && consumePullResult.getPullStatus() == PullStatus.FOUND) {
+                                lagTime = System.currentTimeMillis() - consumePullResult.getMsgFoundList().get(0).getStoreTimestamp();
+                                if (offset.getBrokerOffset() == offset.getConsumerOffset()) {
+                                    lagTime = 0;
+                                }
+                            } else if (consumePullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL) {
+                                PullResult pullResult = ((MQAdminExtImpl) mqAdminExt).queryMsgByOffset(q, consumePullResult.getMinOffset());
+                                if (pullResult != null && pullResult.getPullStatus() == PullStatus.FOUND) {
+                                    lagTime = System.currentTimeMillis() - pullResult.getMsgFoundList().get(0).getStoreTimestamp();
+                                }
                             }
-                        } else if (consumePullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL) {
-                            PullResult pullResult = ((MQAdminExtImpl) mqAdminExt).queryMsgByOffset(q, consumePullResult.getMinOffset());
-                            if (pullResult != null && pullResult.getPullStatus() == PullStatus.FOUND) {
-                                lagTime = System.currentTimeMillis() - pullResult.getMsgFoundList().get(0).getStoreTimestamp();
+                            if (!consumerLatencyMap.containsKey(q.getBrokerName())) {
+                                consumerLatencyMap.put(q.getBrokerName(), lagTime > 0 ? lagTime : 0);
+                            } else if (lagTime > consumerLatencyMap.get(q.getBrokerName())) {
+                                consumerLatencyMap.put(q.getBrokerName(), lagTime);
                             }
                         }
-                        if (!consumerLatencyMap.containsKey(q.getBrokerName())) {
-                            consumerLatencyMap.put(q.getBrokerName(), lagTime > 0 ? lagTime : 0);
-                        } else if (lagTime > consumerLatencyMap.get(q.getBrokerName())) {
-                            consumerLatencyMap.put(q.getBrokerName(), lagTime);
+                        for (Map.Entry<String, Long> consumeLatencyEntry : consumerLatencyMap.entrySet()) {
+                            metricsService.getCollector().addGroupGetLatencyByStoreTimeMetric(clusterName,
+                                    consumeLatencyEntry.getKey(), topic, group, consumeLatencyEntry.getValue());
                         }
-                    }
-                    for (Map.Entry<String, Long> consumeLatencyEntry : consumerLatencyMap.entrySet()) {
-                        metricsService.getCollector().addGroupGetLatencyByStoreTimeMetric(clusterName,
-                            consumeLatencyEntry.getKey(), topic, group, consumeLatencyEntry.getValue());
-                    }
 
-                } catch (Exception ex) {
-                    log.warn("addGroupGetLatencyByStoreTimeMetric error", ex);
+                    } catch (Exception ex) {
+                        log.warn("addGroupGetLatencyByStoreTimeMetric error", ex);
+                    }
                 }
             }
         }