Fix the lag caculation bug for pending consumer group

Author: gnuhpc

docs/index.pdf

@@ -109,5 +109,6 @@
     <orderEntry type="library" name="Maven: commons-io:commons-io:2.5" level="project" />
     <orderEntry type="library" name="Maven: org.apache.commons:commons-lang3:3.5" level="project" />
     <orderEntry type="library" name="Maven: org.apache.curator:curator-framework:2.11.0" level="project" />
+    <orderEntry type="library" name="scala-sdk-2.11.8" level="application" />
   </component>
 </module>

kafka-rest-springboot.iml

@@ -4,14 +4,11 @@
 import kafka.common.OffsetAndMetadata;
 import kafka.coordinator.GroupTopicPartition;
 import lombok.AllArgsConstructor;
-import lombok.NoArgsConstructor;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.gnuhpc.bigdata.constant.ConsumerState;
 import org.gnuhpc.bigdata.constant.ConsumerType;
 import org.gnuhpc.bigdata.utils.KafkaUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
 
 import java.util.Map;
 
@@ -23,27 +20,27 @@ public ConsumerGroupDesc makeOldConsumerGroupDesc(
             Map<Integer, Long> fetchOffSetFromZKResultList,
             String topic, String consumerGroup, TopicMeta topicMeta) {
 
-        ConsumerGroupDesc cgd = ConsumerGroupDesc.newBuilder()
+        ConsumerGroupDesc.Builder cgdBuilder = ConsumerGroupDesc.newBuilder()
                 .setGroupName(consumerGroup)
                 .setTopic(topic)
                 .setPartitionId(op.getKey())
                 .setCurrentOffset(fetchOffSetFromZKResultList.get(op.getKey()))
                 .setLogEndOffset(
                         topicMeta.getTopicPartitionInfos().stream()
-                                .filter(tpi -> tpi.getPartitionId() == op.getKey()).findFirst().get().getEndOffset())
-                .build();
+                                .filter(tpi -> tpi.getPartitionId() == op.getKey()).findFirst().get().getEndOffset());
+
 
         if (op.getValue().equals("none")) {
-            cgd.setConsumerId("-");
-            cgd.setHost("-");
-            cgd.setState(ConsumerState.PENDING);
+            cgdBuilder.setConsumerId("-");
+            cgdBuilder.setHost("-");
+            cgdBuilder.setState(ConsumerState.PENDING);
         } else {
-            cgd.setConsumerId(op.getValue());
-            cgd.setHost(op.getValue().replace(consumerGroup + "_", ""));
-            cgd.setState(ConsumerState.RUNNING);
+            cgdBuilder.setConsumerId(op.getValue());
+            cgdBuilder.setHost(op.getValue().replace(consumerGroup + "_", ""));
+            cgdBuilder.setState(ConsumerState.RUNNING);
         }
-        cgd.setType(ConsumerType.OLD);
-        return cgd;
+        cgdBuilder.setType(ConsumerType.OLD);
+        return cgdBuilder.build();
     }
 
     public ConsumerGroupDesc makeNewRunningConsumerGroupDesc(
@@ -52,31 +49,31 @@ public ConsumerGroupDesc makeNewRunningConsumerGroupDesc(
             Map<Integer, Long> partitionEndOffsetMap,
             AdminClient.ConsumerSummary cs) {
         KafkaConsumer consumer = kafkaUtils.createNewConsumer(consumerGroup);
-        ConsumerGroupDesc cgd = ConsumerGroupDesc.newBuilder()
+        ConsumerGroupDesc.Builder cgdBuilder = ConsumerGroupDesc.newBuilder()
                 .setGroupName(consumerGroup)
                 .setTopic(tp.topic())
                 .setPartitionId(tp.partition())
                 .setConsumerId(cs.clientId())
                 .setHost(cs.clientHost())
                 .setState(ConsumerState.RUNNING)
-                .setType(ConsumerType.NEW).build();
+                .setType(ConsumerType.NEW);
 
         long currentOffset = -1L;
         org.apache.kafka.clients.consumer.OffsetAndMetadata offset = consumer.committed(new TopicPartition(tp.topic(), tp.partition()));
         if (offset != null) {
             currentOffset = offset.offset();
         }
-        cgd.setCurrentOffset(currentOffset);
+        cgdBuilder.setCurrentOffset(currentOffset);
 
         Long endOffset = partitionEndOffsetMap.get(tp.partition());
         if (endOffset == null) { //if endOffset is null ,the partition of this topic has no leader replication
-            cgd.setLogEndOffset(-1l);
+            cgdBuilder.setLogEndOffset(-1l);
         } else {
-            cgd.setLogEndOffset(endOffset);
+            cgdBuilder.setLogEndOffset(endOffset);
         }
         consumer.close();
 
-        return cgd;
+        return cgdBuilder.build();
     }
 
     public ConsumerGroupDesc makeNewPendingConsumerGroupDesc(
@@ -85,26 +82,25 @@ public ConsumerGroupDesc makeNewPendingConsumerGroupDesc(
             Map<Integer, Long> partitionCurrentOffsetMap,
             Map.Entry<GroupTopicPartition, OffsetAndMetadata> storage,
             String topic) {
-        KafkaConsumer consumer = kafkaUtils.createNewConsumer(consumerGroup);
-        ConsumerGroupDesc cgd = ConsumerGroupDesc.newBuilder()
+        int partitionId = storage.getKey().topicPartition().partition();
+        ConsumerGroupDesc.Builder cgdBuilder = ConsumerGroupDesc.newBuilder()
                 .setGroupName(consumerGroup)
                 .setTopic(topic)
                 .setConsumerId("-")
-                .setPartitionId(storage.getKey().topicPartition().partition())
+                .setPartitionId(partitionId)
                 .setCurrentOffset(partitionCurrentOffsetMap.get(storage.getKey().topicPartition().partition()))
                 .setHost("-")
                 .setState(ConsumerState.PENDING)
-                .setType(ConsumerType.NEW)
-                .build();
+                .setType(ConsumerType.NEW);
 
-        Long endOffset = partitionEndOffsetMap.get(cgd.getPartitionId());
+        Long endOffset = partitionEndOffsetMap.get(partitionId);
 
         if (endOffset == null) { //if endOffset is null ,the partition of this topic has no leader replication
-            cgd.setLogEndOffset(-1l);
+            cgdBuilder.setLogEndOffset(-1l);
         } else {
-            cgd.setLogEndOffset(endOffset);
+            cgdBuilder.setLogEndOffset(endOffset);
         }
 
-        return cgd;
+        return cgdBuilder.build();
     }
 }

src/main/java/org/gnuhpc/bigdata/model/ConsumerGroupDescFactory.java

@@ -648,14 +648,14 @@ private List<ConsumerGroupDesc> setPendingNewCGD(
         ConsumerGroupDescFactory factory = new ConsumerGroupDescFactory(kafkaUtils);
         cgdList.addAll(
                 topicStorage.entrySet().stream().map(
-                        storage-> factory.makeNewPendingConsumerGroupDesc(
+                        storage -> factory.makeNewPendingConsumerGroupDesc(
                                 consumerGroup,
                                 partitionEndOffsetMap,
                                 partitionCurrentOffsetMap,
                                 storage,
                                 topic)
                 )
-                .collect(toList()));
+                        .collect(toList()));
 
         return cgdList;
     }