class DynamicBrokersReader {

    public GlobalPartitionInformation getBrokerInfo() throws SocketTimeoutException {
      GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation();
        try {
            int numPartitionsForTopic = getNumPartitions();
            String brokerInfoPath = brokerPath();
            for (int partition = 0; partition < numPartitionsForTopic; partition++) {
                //获取每个分区的leader节点的编号
                int leader = getLeaderFor(partition);
                //在zk上的leader节点
                String path = brokerInfoPath + "/" + leader;
                try {
                    //leader节点的数据
                    byte[] brokerData = _curator.getData().forPath(path);
                    //重构Broker对象, 即通过配置在ZK上的配置信息(ZK作为注册中心), 获取Kafka集群的Brokers列表
                    Broker hp = getBrokerHost(brokerData);
                    //每个分区, Broker
                    globalPartitionInformation.addPartition(partition, hp);
                } catch (org.apache.zookeeper.KeeperException.NoNodeException e) {
                    LOG.error("Node {} does not exist ", path);
                }
            }
        } catch (SocketTimeoutException e) {
					throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        LOG.info("Read partition info from zookeeper: " + globalPartitionInformation);
        return globalPartitionInformation;
    }

    public  DynamicBrokersReader(Map conf, String zkStr, String zkPath, String topic);

}