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

修复因DB中Broker信息不存在导致TotalLogSize指标获取时抛空指针问题

Browse Source
This commit is contained in:
zengqiao
2022-09-20 10:27:30 +08:00
parent 375c6f56c9
commit e06712397e
2 changed files with 12 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterMetricServiceImpl.java
View File

@@ -751,8 +751,8 @@ public class ClusterMetricServiceImpl extends BaseMetricService implements Clust
private Result<ClusterMetrics> getMetricFromKafkaByTotalTopics(Long clusterId, String metric, String topicMetric){ private Result<ClusterMetrics> getMetricFromKafkaByTotalTopics(Long clusterId, String metric, String topicMetric){
List<Topic> topics = topicService.listTopicsFromCacheFirst(clusterId); List<Topic> topics = topicService.listTopicsFromCacheFirst(clusterId);
float metricsSum = 0f; float sumMetricValue = 0f;
for(Topic topic : topics){ for(Topic topic : topics) {
Result<List<TopicMetrics>> ret = topicMetricService.collectTopicMetricsFromKafkaWithCacheFirst( Result<List<TopicMetrics>> ret = topicMetricService.collectTopicMetricsFromKafkaWithCacheFirst(
clusterId, clusterId,
topic.getTopicName(), topic.getTopicName(),
@@ -763,14 +763,15 @@ public class ClusterMetricServiceImpl extends BaseMetricService implements Clust
continue; continue;
} }
List<TopicMetrics> topicMetrics = ret.getData(); for (TopicMetrics metrics : ret.getData()) {
for (TopicMetrics metrics : topicMetrics) { if(metrics.isBBrokerAgg()) {
if(metrics.isBBrokerAgg()){ Float metricValue = metrics.getMetric(topicMetric);
metricsSum += Double.valueOf(metrics.getMetrics().get(topicMetric)); sumMetricValue += (metricValue == null? 0f: metricValue);
break;
} }
} }
} }
return Result.buildSuc(initWithMetrics(clusterId, metric, metricsSum)); return Result.buildSuc(initWithMetrics(clusterId, metric, sumMetricValue));
} }
} }

4
km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaJMXClient.java
View File

@@ -191,6 +191,10 @@ public class KafkaJMXClient extends AbstractClusterLoadedChangedHandler {
lambdaQueryWrapper.eq(BrokerPO::getStatus, Constant.ALIVE); lambdaQueryWrapper.eq(BrokerPO::getStatus, Constant.ALIVE);
BrokerPO brokerPO = brokerDAO.selectOne(lambdaQueryWrapper); BrokerPO brokerPO = brokerDAO.selectOne(lambdaQueryWrapper);
if (brokerPO == null) {
return null;
}
return Broker.buildFrom(brokerPO); return Broker.buildFrom(brokerPO);
} }
} }