1、调整KafkaZKDao位置；2、offset信息获取时，过滤掉无leader分区；3、调整验证ZK是否合法时的session超时时间

zengqiao · zengqiao · commit 725ac10c3d88 · 2022-09-22T11:30:46.000+08:00
diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/broker/Broker.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/broker/Broker.java
@@ -5,7 +5,6 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.common.IpPortData;
 import com.xiaojukeji.know.streaming.km.common.bean.po.broker.BrokerPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.BrokerMetadata;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -79,20 +78,6 @@ public static Broker buildFrom(Long clusterPhyId, Node node, Long startTimestamp
         return metadata;
     }
 
-    public static Broker buildFrom(Long clusterPhyId, Integer brokerId, BrokerMetadata brokerMetadata) {
-        Broker metadata = new Broker();
-        metadata.setClusterPhyId(clusterPhyId);
-        metadata.setBrokerId(brokerId);
-        metadata.setHost(brokerMetadata.getHost());
-        metadata.setPort(brokerMetadata.getPort());
-        metadata.setJmxPort(brokerMetadata.getJmxPort());
-        metadata.setStartTimestamp(brokerMetadata.getTimestamp());
-        metadata.setRack(brokerMetadata.getRack());
-        metadata.setStatus(1);
-        metadata.setEndpointMap(brokerMetadata.getEndpointMap());
-        return metadata;
-    }
-
     public static Broker buildFrom(BrokerPO brokerPO) {
         Broker broker = ConvertUtil.obj2Obj(brokerPO, Broker.class);
         String endpointMapStr = brokerPO.getEndpointMap();
diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/MsgConstant.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/MsgConstant.java
@@ -52,6 +52,10 @@ public static String getClusterTopicKey(Long clusterPhyId, String topicName) {
 
     /**************************************************** Partition ****************************************************/
 
+    public static String getPartitionNoLeader(Long clusterPhyId, String topicName) {
+        return String.format("集群ID:[%d] Topic名称:[%s] 所有分区NoLeader", clusterPhyId, topicName);
+    }
+
     public static String getPartitionNotExist(Long clusterPhyId, String topicName) {
         return String.format("集群ID:[%d] Topic名称:[%s] 存在非法的分区ID", clusterPhyId, topicName);
     }
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/AbstractZKWatcher.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/AbstractZKWatcher.java
@@ -5,7 +5,7 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.KafkaZkClient;
 import org.springframework.beans.factory.annotation.Autowired;
 
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/AbstractZKHandler.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/AbstractZKHandler.java
@@ -7,7 +7,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.change.record.KafkaChangeRecordService;
 import com.xiaojukeji.know.streaming.km.persistence.cache.LoadedClusterPhyCache;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 
 
 public abstract class AbstractZKHandler {
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/BrokersNodeChangeHandler.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/BrokersNodeChangeHandler.java
@@ -9,7 +9,7 @@
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
 import com.xiaojukeji.know.streaming.km.core.service.change.record.KafkaChangeRecordService;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.BrokerIdsZNode;
 import kafka.zookeeper.ZNodeChildChangeHandler;
 
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/ConfigNotificationNodeChangeHandler.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/ConfigNotificationNodeChangeHandler.java
@@ -8,11 +8,11 @@
 import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.OperationEnum;
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config.ConfigChangeNotificationBaseData;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config.ConfigChangeNotificationDataV1;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config.ConfigChangeNotificationDataV2;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config.ConfigChangeNotificationBaseData;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config.ConfigChangeNotificationDataV1;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config.ConfigChangeNotificationDataV2;
 import com.xiaojukeji.know.streaming.km.core.service.change.record.KafkaChangeRecordService;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.ConfigEntityChangeNotificationZNode;
 import kafka.zookeeper.ZNodeChildChangeHandler;
 import org.apache.zookeeper.data.Stat;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/ControllerNodeChangeHandler.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/ControllerNodeChangeHandler.java
@@ -11,7 +11,7 @@
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
 import com.xiaojukeji.know.streaming.km.core.service.change.record.KafkaChangeRecordService;
 import com.xiaojukeji.know.streaming.km.core.service.kafkacontroller.KafkaControllerService;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.ControllerZNode;
 import kafka.zookeeper.ZNodeChangeHandler;
 
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/TopicsNodeChangeHandler.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/flusher/zk/handler/TopicsNodeChangeHandler.java
@@ -9,7 +9,7 @@
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
 import com.xiaojukeji.know.streaming.km.core.service.change.record.KafkaChangeRecordService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.TopicsZNode;
 import kafka.zookeeper.ZNodeChildChangeHandler;
 
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java
@@ -24,16 +24,14 @@
 import com.xiaojukeji.know.streaming.km.common.jmx.JmxConnectorWrap;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.BrokerMetadata;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
 import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
 import com.xiaojukeji.know.streaming.km.persistence.jmx.JmxDAO;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaJMXClient;
 import com.xiaojukeji.know.streaming.km.persistence.mysql.broker.BrokerDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
-import kafka.zk.BrokerIdZNode;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.BrokerIdsZNode;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
@@ -310,9 +308,7 @@ private Result<List<Broker>> getBrokersFromZKClient(ClusterPhy clusterPhy) {
 
             List<String> brokerIdList = kafkaZKDAO.getChildren(clusterPhy.getId(), BrokerIdsZNode.path(), false);
             for (String brokerId: brokerIdList) {
-                BrokerMetadata metadata = kafkaZKDAO.getData(clusterPhy.getId(), BrokerIdZNode.path(Integer.valueOf(brokerId)), BrokerMetadata.class);
-                BrokerMetadata.parseAndUpdateBrokerMetadata(metadata);
-                brokerList.add(Broker.buildFrom(clusterPhy.getId(), Integer.valueOf(brokerId), metadata));
+                brokerList.add(kafkaZKDAO.getBrokerMetadata(clusterPhy.getId(), Integer.valueOf(brokerId)));
             }
 
             return Result.buildSuc(brokerList);
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterValidateServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterValidateServiceImpl.java
@@ -13,8 +13,8 @@
 import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterValidateService;
 import com.xiaojukeji.know.streaming.km.persistence.jmx.JmxDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.impl.KafkaZKDAOImpl;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.impl.KafkaZKDAOImpl;
 import kafka.server.KafkaConfig;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.*;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/kafkacontroller/impl/KafkaControllerServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/kafkacontroller/impl/KafkaControllerServiceImpl.java
@@ -19,7 +19,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.kafkacontroller.KafkaControllerService;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.mysql.kafkacontroller.KafkaControllerDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
 import org.springframework.beans.factory.annotation.Autowired;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/PartitionServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/PartitionServiceImpl.java
@@ -21,14 +21,14 @@
 import com.xiaojukeji.know.streaming.km.common.exception.VCHandlerNotExistException;
 import com.xiaojukeji.know.streaming.km.common.utils.CommonUtils;
 import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.PartitionMap;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.PartitionState;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers.PartitionMap;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers.PartitionState;
 import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService;
 import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaConsumerClient;
 import com.xiaojukeji.know.streaming.km.persistence.mysql.partition.PartitionDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.TopicPartitionStateZNode;
 import kafka.zk.TopicPartitionsZNode;
 import kafka.zk.TopicZNode;
@@ -202,10 +202,22 @@ public Integer getNoLeaderPartitionSizeByClusterId(Long clusterPhyId) {
     @Override
     public Result<Map<TopicPartition, Long>> getPartitionOffsetFromKafka(Long clusterPhyId, String topicName, OffsetSpec offsetSpec, Long timestamp) {
         Map<TopicPartition, OffsetSpec> topicPartitionOffsets = new HashMap<>();
-        this.listPartitionByTopic(clusterPhyId, topicName)
-                .stream()
+
+        List<Partition> partitionList = this.listPartitionByTopic(clusterPhyId, topicName);
+        if (partitionList == null || partitionList.isEmpty()) {
+            // Topic不存在
+            return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getTopicNotExist(clusterPhyId, topicName));
+        }
+
+        partitionList.stream()
+                .filter(item -> !item.getLeaderBrokerId().equals(KafkaConstant.NO_LEADER))
                 .forEach(elem -> topicPartitionOffsets.put(new TopicPartition(topicName, elem.getPartitionId()), offsetSpec));
 
+        if (topicPartitionOffsets.isEmpty()) {
+            // 所有分区no-leader
+            return Result.buildFromRSAndMsg(ResultStatus.OPERATION_FAILED, MsgConstant.getPartitionNoLeader(clusterPhyId, topicName));
+        }
+
         try {
             return (Result<Map<TopicPartition, Long>>) doVCHandler(clusterPhyId, PARTITION_OFFSET_GET, new PartitionOffsetParam(clusterPhyId, topicName, topicPartitionOffsets, timestamp));
         } catch (VCHandlerNotExistException e) {
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/OpTopicServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/OpTopicServiceImpl.java
@@ -23,7 +23,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.controller.ReplicaAssignment;
 import kafka.server.ConfigType;
 import kafka.zk.AdminZkClient;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/TopicConfigServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/TopicConfigServiceImpl.java
@@ -30,7 +30,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.server.ConfigType;
 import kafka.zk.AdminZkClient;
 import kafka.zk.KafkaZkClient;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/TopicServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/TopicServiceImpl.java
@@ -23,7 +23,7 @@
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
 import com.xiaojukeji.know.streaming.km.persistence.mysql.topic.TopicDAO;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.zk.TopicsZNode;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.TopicPartitionInfo;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/package-info.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * 读取Kafka在ZK中存储的数据的包
+ */
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/service/KafkaZKDAO.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/service/KafkaZKDAO.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.persistence.zk;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service;
 
 import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.kafkacontroller.KafkaController;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/service/impl/KafkaZKDAOImpl.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/service/impl/KafkaZKDAOImpl.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.persistence.zk.impl;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.impl;
 
 import com.alibaba.fastjson.JSON;
 import com.didiglobal.logi.log.ILog;
@@ -11,11 +11,11 @@
 import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException;
 import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
 import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.ControllerData;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.BrokerMetadata;
-import com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers.PartitionMap;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.ControllerData;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers.BrokerMetadata;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers.PartitionMap;
 import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
-import com.xiaojukeji.know.streaming.km.persistence.zk.KafkaZKDAO;
+import com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.service.KafkaZKDAO;
 import kafka.utils.Json;
 import kafka.zk.*;
 import kafka.zookeeper.AsyncResponse;
@@ -46,14 +46,14 @@ public class KafkaZKDAOImpl implements KafkaZKDAO {
     public Broker getBrokerMetadata(String zkAddress) throws KeeperException.NoNodeException, AdminOperateException {
         ZooKeeper zooKeeper = null;
         try {
-            zooKeeper = new ZooKeeper(zkAddress, 1000, watchedEvent -> logger.info(" receive event : " + watchedEvent.getType().name()));
+            zooKeeper = new ZooKeeper(zkAddress, 3000, watchedEvent -> logger.info(" receive event : " + watchedEvent.getType().name()));
             List<String> brokerIdList = this.getChildren(zooKeeper, BrokerIdsZNode.path());
             if (brokerIdList == null || brokerIdList.isEmpty()) {
                 return null;
             }
 
             BrokerMetadata brokerMetadata = this.getData(zooKeeper, BrokerIdZNode.path(Integer.parseInt(brokerIdList.get(0))), false, BrokerMetadata.class);
-            return Broker.buildFrom(null, Integer.valueOf(brokerIdList.get(0)), brokerMetadata);
+            return this.convert2Broker(null, Integer.valueOf(brokerIdList.get(0)), brokerMetadata);
         }  catch (KeeperException.NoNodeException nne) {
             logger.warn("method=getBrokerMetadata||zkAddress={}||errMsg=exception", zkAddress, nne);
             throw nne;
@@ -79,7 +79,7 @@ public Broker getBrokerMetadata(Long clusterPhyId, Integer brokerId) throws NotE
         try {
             BrokerMetadata metadata = this.getData(kafkaZkClient.currentZooKeeper(), BrokerIdZNode.path(brokerId), false, BrokerMetadata.class);
             BrokerMetadata.parseAndUpdateBrokerMetadata(metadata);
-            return Broker.buildFrom(clusterPhyId, brokerId, metadata);
+            return this.convert2Broker(clusterPhyId, brokerId, metadata);
         } catch (KeeperException ke) {
             logger.error("method=getBrokerMetadata||clusterPhyId={}||brokerId={}||errMsg=exception", clusterPhyId, brokerId, ke);
             throw ke;
@@ -269,4 +269,18 @@ private <T> T getData(ZooKeeper zooKeeper, String path, boolean addWatch, Class<
         byte[] bytes = zooKeeper.getData(path, addWatch, null);
         return JSON.parseObject(bytes, clazz);
     }
+
+    private Broker convert2Broker(Long clusterPhyId, Integer brokerId, BrokerMetadata brokerMetadata) {
+        Broker metadata = new Broker();
+        metadata.setClusterPhyId(clusterPhyId);
+        metadata.setBrokerId(brokerId);
+        metadata.setHost(brokerMetadata.getHost());
+        metadata.setPort(brokerMetadata.getPort());
+        metadata.setJmxPort(brokerMetadata.getJmxPort());
+        metadata.setStartTimestamp(brokerMetadata.getTimestamp());
+        metadata.setRack(brokerMetadata.getRack());
+        metadata.setStatus(1);
+        metadata.setEndpointMap(brokerMetadata.getEndpointMap());
+        return metadata;
+    }
 }
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/ControllerData.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/ControllerData.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/BrokerMetadata.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/BrokerMetadata.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/PartitionMap.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/PartitionMap.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/PartitionState.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/PartitionState.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.AllArgsConstructor;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/TopicMetadata.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/brokers/TopicMetadata.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.brokers;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.brokers;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationBaseData.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationBaseData.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationDataV1.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationDataV1.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.AllArgsConstructor;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationDataV2.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigChangeNotificationDataV2.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.AllArgsConstructor;
diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigNodeData.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/zookeeper/znode/config/ConfigNodeData.java
@@ -1,4 +1,4 @@
-package com.xiaojukeji.know.streaming.km.common.zookeeper.znode.config;
+package com.xiaojukeji.know.streaming.km.persistence.kafka.zookeeper.znode.config;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;

Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,10 @@ public static String getClusterTopicKey(Long clusterPhyId, String topicName) {`
`52`	`52`
`53`	`53`	`/************************************************** Partition **************************************************/`
`54`	`54`
	`55`	`+ public static String getPartitionNoLeader(Long clusterPhyId, String topicName) {`
	`56`	`+ return String.format("集群ID:[%d] Topic名称:[%s] 所有分区NoLeader", clusterPhyId, topicName);`
	`57`	`+ }`
	`58`	`+`
`55`	`59`	`public static String getPartitionNotExist(Long clusterPhyId, String topicName) {`
`56`	`60`	`return String.format("集群ID:[%d] Topic名称:[%s] 存在非法的分区ID", clusterPhyId, topicName);`
`57`	`61`	`}`