import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
/**
 * Generates the Kafka bootstrap connection string from the metadata stored in Zookeeper.
 * Allows for backwards compatibility of the zookeeperConnect configuration.
 */
private String lookupBootstrap(String zookeeperConnect, SecurityProtocol securityProtocol) {
  ZkUtils zkUtils = ZkUtils.apply(zookeeperConnect, ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT,
      JaasUtils.isZkSecurityEnabled());
  try {
    List<BrokerEndPoint> endPoints =
        asJavaListConverter(zkUtils.getAllBrokerEndPointsForChannel(securityProtocol)).asJava();
    List<String> connections = new ArrayList<>();
    for (BrokerEndPoint endPoint : endPoints) {
      connections.add(endPoint.connectionString());
    }
    return StringUtils.join(connections, ',');
  } finally {
    zkUtils.close();
  }
}
 

import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
public MockKafkaSimpleConsumerFactory(String[] hosts, int[] ports, long[] partitionStartOffsets,
    long[] partitionEndOffsets, int[] partitionLeaderIndices, String topicName) {
  Preconditions.checkArgument(hosts.length == ports.length);
  this.hosts = hosts;
  this.ports = ports;
  brokerCount = hosts.length;

  brokerArray = new BrokerEndPoint[brokerCount];
  for (int i = 0; i < brokerCount; i++) {
    brokerArray[i] = new BrokerEndPoint(i, hosts[i], ports[i]);
  }

  Preconditions.checkArgument(partitionStartOffsets.length == partitionEndOffsets.length);
  Preconditions.checkArgument(partitionStartOffsets.length == partitionLeaderIndices.length);
  this.partitionStartOffsets = partitionStartOffsets;
  this.partitionEndOffsets = partitionEndOffsets;
  this.partitionLeaderIndices = partitionLeaderIndices;
  partitionCount = partitionStartOffsets.length;

  this.topicName = topicName;
}
 

import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
/**
 * 获取指定主题及分区logsize
 * @param stat
 */
public void getTopicPartitionLogSize(TopicPartitionInfo stat) {
    BrokerEndPoint leader = findLeader(stat.getTopic(), stat.getPartition()).leader();
    SimpleConsumer consumer = getConsumerClient(leader.host(), leader.port());

    try {
        long logsize = getLastOffset(consumer, stat.getTopic(), stat.getPartition(),
            kafka.api.OffsetRequest.LatestTime());
        stat.setLogSize(logsize);
    } finally {
        consumer.close();
    }
}
 

import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
/**
   * 获取指定主题及分区logsize
   * @param stat
   */
  public void getTopicPartitionLogSize(TopicPartitionInfo stat){
  	BrokerEndPoint leader = findLeader(stat.getTopic(), stat.getPartition()).leader();
  	SimpleConsumer consumer = getConsumerClient(leader.host(), leader.port());	
  	
  	try {			
  		long logsize = getLastOffset(consumer,stat.getTopic(), stat.getPartition(), kafka.api.OffsetRequest.LatestTime());
  		stat.setLogSize(logsize);
} finally {
	consumer.close();
}
  }
 

import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
@Override
public TopicMetadataResponse send(TopicMetadataRequest request) {
  java.util.List<String> topics = request.topics();
  TopicMetadata[] topicMetadataArray = new TopicMetadata[topics.size()];

  for (int i = 0; i < topicMetadataArray.length; i++) {
    String topic = topics.get(i);
    if (!topic.equals(topicName)) {
      topicMetadataArray[i] = new TopicMetadata(topic, null, Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
    } else {
      PartitionMetadata[] partitionMetadataArray = new PartitionMetadata[partitionCount];
      for (int j = 0; j < partitionCount; j++) {
        java.util.List<BrokerEndPoint> emptyJavaList = Collections.emptyList();
        List<BrokerEndPoint> emptyScalaList = JavaConversions.asScalaBuffer(emptyJavaList).toList();
        partitionMetadataArray[j] = new PartitionMetadata(j, Some.apply(brokerArray[partitionLeaderIndices[j]]),
            emptyScalaList, emptyScalaList, Errors.NONE.code());
      }

      Seq<PartitionMetadata> partitionsMetadata = List.fromArray(partitionMetadataArray);
      topicMetadataArray[i] = new TopicMetadata(topic, partitionsMetadata, Errors.NONE.code());
    }
  }

  Seq<BrokerEndPoint> brokers = List.fromArray(brokerArray);
  Seq<TopicMetadata> topicsMetadata = List.fromArray(topicMetadataArray);

  return new TopicMetadataResponse(new kafka.api.TopicMetadataResponse(brokers, topicsMetadata, -1));
}
 

import kafka.cluster.BrokerEndPoint; //导入依赖的package包/类
private static String getHostPort(BrokerEndPoint leader) {
  if (leader != null) {
    return leader.host() + ":" + leader.port();
  }
  return null;
}