bugfix, fix collect consumer metrics task

Author: zengqiao

common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/dto/consumer/ConsumerDTO.java

@@ -1,8 +1,5 @@
 package com.xiaojukeji.kafka.manager.common.entity.dto.consumer;
 
-import com.xiaojukeji.kafka.manager.common.entity.zookeeper.PartitionState;
-
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -11,20 +8,33 @@
  * @date 2015/11/12
  */
 public class ConsumerDTO {
-    /**
-     * 消费group名
-     */
+    private Long clusterId;
+
+    private String topicName;
+
     private String consumerGroup;
 
-    /**
-     * 消费类型，一般为static
-     */
     private String location;
 
-    /**
-     * 订阅的每个topic的partition状态列表
-     */
-    private Map<String, List<PartitionState>> topicPartitionMap;
+    private Map<Integer, Long> partitionOffsetMap;
+
+    private Map<Integer, Long> consumerOffsetMap;
+
+    public Long getClusterId() {
+        return clusterId;
+    }
+
+    public void setClusterId(Long clusterId) {
+        this.clusterId = clusterId;
+    }
+
+    public String getTopicName() {
+        return topicName;
+    }
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
 
     public String getConsumerGroup() {
         return consumerGroup;
@@ -42,20 +52,31 @@ public void setLocation(String location) {
         this.location = location;
     }
 
-    public Map<String, List<PartitionState>> getTopicPartitionMap() {
-        return topicPartitionMap;
+    public Map<Integer, Long> getPartitionOffsetMap() {
+        return partitionOffsetMap;
+    }
+
+    public void setPartitionOffsetMap(Map<Integer, Long> partitionOffsetMap) {
+        this.partitionOffsetMap = partitionOffsetMap;
+    }
+
+    public Map<Integer, Long> getConsumerOffsetMap() {
+        return consumerOffsetMap;
     }
 
-    public void setTopicPartitionMap(Map<String, List<PartitionState>> topicPartitionMap) {
-        this.topicPartitionMap = topicPartitionMap;
+    public void setConsumerOffsetMap(Map<Integer, Long> consumerOffsetMap) {
+        this.consumerOffsetMap = consumerOffsetMap;
     }
 
     @Override
     public String toString() {
-        return "Consumer{" +
-                "consumerGroup='" + consumerGroup + '\'' +
+        return "ConsumerDTO{" +
+                "clusterId=" + clusterId +
+                ", topicName='" + topicName + '\'' +
+                ", consumerGroup='" + consumerGroup + '\'' +
                 ", location='" + location + '\'' +
-                ", topicPartitionMap=" + topicPartitionMap +
+                ", partitionOffsetMap=" + partitionOffsetMap +
+                ", consumerOffsetMap=" + consumerOffsetMap +
                 '}';
     }
 }

service/src/main/java/com/xiaojukeji/kafka/manager/service/collector/CollectConsumerMetricsTask.java

@@ -2,12 +2,12 @@
 
 import com.xiaojukeji.kafka.manager.common.constant.Constant;
 import com.xiaojukeji.kafka.manager.common.entity.ConsumerMetrics;
-import com.xiaojukeji.kafka.manager.common.entity.zookeeper.PartitionState;
 import com.xiaojukeji.kafka.manager.common.entity.dto.consumer.ConsumerDTO;
 import com.xiaojukeji.kafka.manager.common.entity.po.ClusterDO;
 import com.xiaojukeji.kafka.manager.service.cache.ClusterMetadataManager;
 import com.xiaojukeji.kafka.manager.service.cache.KafkaMetricsCache;
 import com.xiaojukeji.kafka.manager.service.service.ConsumerService;
+import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,8 +34,8 @@ public void collect() {
         if (clusterDO == null) {
             return;
         }
-        Map<String, List<PartitionState>> topicNamePartitionStateListMap = new HashMap<>();
-        List<ConsumerDTO> consumerDTOList = consumerService.getMonitoredConsumerList(clusterDO, topicNamePartitionStateListMap);
+        Map<TopicPartition, Long> allPartitionOffsetMap = new HashMap<>();
+        List<ConsumerDTO> consumerDTOList = consumerService.getMonitoredConsumerList(clusterDO, allPartitionOffsetMap);
 
         List<ConsumerMetrics> consumerMetricsList = convert2ConsumerMetrics(consumerDTOList);
         KafkaMetricsCache.putConsumerMetricsToCache(clusterId, consumerMetricsList);
@@ -47,23 +47,27 @@ public void collect() {
     private List<ConsumerMetrics> convert2ConsumerMetrics(List<ConsumerDTO> consumerDTOList) {
         List<ConsumerMetrics> consumerMetricsList = new ArrayList<>();
         for (ConsumerDTO consumerDTO : consumerDTOList) {
-            Map<String, List<PartitionState>> topicNamePartitionStateListMap = consumerDTO.getTopicPartitionMap();
-            for(Map.Entry<String, List<PartitionState>> entry : topicNamePartitionStateListMap.entrySet()){
-                String topicName = entry.getKey();
-                List<PartitionState> partitionStateList = entry.getValue();
-                ConsumerMetrics consumerMetrics = new ConsumerMetrics();
-                consumerMetrics.setClusterId(clusterId);
-                consumerMetrics.setConsumerGroup(consumerDTO.getConsumerGroup());
-                consumerMetrics.setLocation(consumerDTO.getLocation());
-                consumerMetrics.setTopicName(topicName);
-                long sumLag = 0;
-                for (PartitionState partitionState : partitionStateList) {
-                    Map.Entry<Long, Long> offsetEntry = new AbstractMap.SimpleEntry<>(partitionState.getOffset(), partitionState.getConsumeOffset());
-                    sumLag += (offsetEntry.getKey() - offsetEntry.getValue() > 0 ? offsetEntry.getKey() - offsetEntry.getValue(): 0);
+            if (consumerDTO.getPartitionOffsetMap() == null || consumerDTO.getConsumerOffsetMap() == null) {
+                continue;
+            }
+
+            ConsumerMetrics consumerMetrics = new ConsumerMetrics();
+            consumerMetrics.setClusterId(consumerDTO.getClusterId());
+            consumerMetrics.setConsumerGroup(consumerDTO.getConsumerGroup());
+            consumerMetrics.setLocation(consumerDTO.getLocation());
+            consumerMetrics.setTopicName(consumerDTO.getTopicName());
+
+            long sumLag = 0;
+            for(Map.Entry<Integer, Long> entry : consumerDTO.getPartitionOffsetMap().entrySet()){
+                Long partitionOffset = entry.getValue();
+                Long consumerOffset = consumerDTO.getConsumerOffsetMap().get(entry.getKey());
+                if (partitionOffset == null || consumerOffset == null) {
+                    continue;
                 }
-                consumerMetrics.setSumLag(sumLag);
-                consumerMetricsList.add(consumerMetrics);
+                sumLag += Math.max(partitionOffset - consumerOffset, 0);
             }
+            consumerMetrics.setSumLag(sumLag);
+            consumerMetricsList.add(consumerMetrics);
         }
         return consumerMetricsList;
     }

service/src/main/java/com/xiaojukeji/kafka/manager/service/service/ConsumerService.java

@@ -7,6 +7,7 @@
 import com.xiaojukeji.kafka.manager.common.entity.dto.PartitionOffsetDTO;
 import com.xiaojukeji.kafka.manager.common.entity.dto.consumer.ConsumeDetailDTO;
 import com.xiaojukeji.kafka.manager.common.entity.po.ClusterDO;
+import org.apache.kafka.common.TopicPartition;
 
 import java.util.List;
 import java.util.Map;
@@ -57,7 +58,7 @@ public interface ConsumerService {
      * @return
      */
     List<ConsumerDTO> getMonitoredConsumerList(ClusterDO clusterDO,
-                                               Map<String, List<PartitionState>> topicNamePartitionStateListMap);
+                                               Map<TopicPartition, Long> partitionOffsetMap);
 
     /**
      * 重置offset