From a5fa9de54b04a8edd904ec37fd7d1cdf3dbd6bfb Mon Sep 17 00:00:00 2001
From: zengqiao <zengqiao@didiglobal.com>
Date: Wed, 28 Sep 2022 19:52:11 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DGroup=E6=8C=87=E6=A0=87?=
 =?UTF-8?q?=E9=98=B2=E9=87=8D=E5=A4=8D=E4=B8=8D=E7=94=9F=E6=95=88=E9=97=AE?=
 =?UTF-8?q?=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../group/impl/GroupMetricServiceImpl.java    | 26 ++++++++++++-------
 1 file changed, 17 insertions(+), 9 deletions(-)

diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
index ea324888..427edc2c 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
@@ -90,23 +90,31 @@ public class GroupMetricServiceImpl extends BaseMetricService implements GroupMe
 
     @Override
     public Result<List<GroupMetrics>> collectGroupMetricsFromKafka(Long clusterId, String groupName, List<String> metrics) {
-        List<GroupMetrics>          allGroupMetrics         = new ArrayList<>();
-        Map<String, GroupMetrics>   topicPartitionGroupMap  = new HashMap<>();
+        List<GroupMetrics> allGroupMetrics = new ArrayList<>();
+        Map<String, GroupMetrics> topicPartitionGroupMap = new HashMap<>();
 
         GroupMetrics groupMetrics = new GroupMetrics(clusterId, groupName, true);
-        for(String metric : metrics){
-            if(null != groupMetrics.getMetrics().get(metric)){continue;}
+        Set<String> existMetricSet = new HashSet<>();
+        for (String metric : metrics) {
+            if (existMetricSet.contains(metric)) {
+                continue;
+            }
 
             Result<List<GroupMetrics>> ret = collectGroupMetricsFromKafka(clusterId, groupName, metric);
-            if(null != ret && ret.successful()){
+            if (null != ret && ret.successful()) {
                 List<GroupMetrics> groupMetricsList = ret.getData();
-                for(GroupMetrics gm : groupMetricsList){
-                    if(gm.isBGroupMetric()){
+
+                for (GroupMetrics gm : groupMetricsList) {
+
+                    //记录已存在的指标
+                    existMetricSet.addAll(gm.getMetrics().keySet());
+
+                    if (gm.isBGroupMetric()) {
                         groupMetrics.getMetrics().putAll(gm.getMetrics());
-                    }else {
+                    } else {
                         GroupMetrics topicGroupMetric = topicPartitionGroupMap.getOrDefault(
                                 gm.getTopic() + gm.getPartitionId(),
-                                     new GroupMetrics(clusterId, groupName, false));
+                                new GroupMetrics(clusterId, gm.getPartitionId(), gm.getTopic(), groupName, false));
 
                         topicGroupMetric.getMetrics().putAll(gm.getMetrics());
                         topicPartitionGroupMap.put(gm.getTopic() + gm.getPartitionId(), topicGroupMetric);