8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2026-01-02 02:02:13 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

1、修复Group指标防重复不生效问题；2、修复自动创建ES索引模版失败问题； (#628)

Browse Source 
* 修复自动创建ES索引模版失败问题

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

Co-authored-by: zengqiao <zengqiao@didiglobal.com>
This commit is contained in:
EricZeng
2022-09-28 19:55:30 +08:00
committed by GitHub
parent 95f1a2c630
commit 7815126ff5
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
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);