8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

修复Group指标防重复不生效问题

Browse Source
This commit is contained in:
zengqiao
2022-09-28 19:52:11 +08:00
parent 1e256ae1fd
commit a5fa9de54b
1 changed files with 17 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupMetricServiceImpl.java
View File

@@ -90,23 +90,31 @@ public class GroupMetricServiceImpl extends BaseMetricService implements GroupMe
@Override @Override
public Result<List<GroupMetrics>> collectGroupMetricsFromKafka(Long clusterId, String groupName, List<String> metrics) { public Result<List<GroupMetrics>> collectGroupMetricsFromKafka(Long clusterId, String groupName, List<String> metrics) {
List<GroupMetrics> allGroupMetrics = new ArrayList<>(); List<GroupMetrics> allGroupMetrics = new ArrayList<>();
Map<String, GroupMetrics> topicPartitionGroupMap = new HashMap<>(); Map<String, GroupMetrics> topicPartitionGroupMap = new HashMap<>();
GroupMetrics groupMetrics = new GroupMetrics(clusterId, groupName, true); GroupMetrics groupMetrics = new GroupMetrics(clusterId, groupName, true);
for(String metric : metrics){ Set<String> existMetricSet = new HashSet<>();
if(null != groupMetrics.getMetrics().get(metric)){continue;} for (String metric : metrics) {
if (existMetricSet.contains(metric)) {
continue;
}
Result<List<GroupMetrics>> ret = collectGroupMetricsFromKafka(clusterId, groupName, metric); Result<List<GroupMetrics>> ret = collectGroupMetricsFromKafka(clusterId, groupName, metric);
if(null != ret && ret.successful()){ if (null != ret && ret.successful()) {
List<GroupMetrics> groupMetricsList = ret.getData(); 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()); groupMetrics.getMetrics().putAll(gm.getMetrics());
}else { } else {
GroupMetrics topicGroupMetric = topicPartitionGroupMap.getOrDefault( GroupMetrics topicGroupMetric = topicPartitionGroupMap.getOrDefault(
gm.getTopic() + gm.getPartitionId(), gm.getTopic() + gm.getPartitionId(),
new GroupMetrics(clusterId, groupName, false)); new GroupMetrics(clusterId, gm.getPartitionId(), gm.getTopic(), groupName, false));
topicGroupMetric.getMetrics().putAll(gm.getMetrics()); topicGroupMetric.getMetrics().putAll(gm.getMetrics());
topicPartitionGroupMap.put(gm.getTopic() + gm.getPartitionId(), topicGroupMetric); topicPartitionGroupMap.put(gm.getTopic() + gm.getPartitionId(), topicGroupMetric);