test_class.declarator: dynamicBrokersReader
test_class.modifier: private
test_class.original_string: private DynamicBrokersReader dynamicBrokersReader;
test_class.type: DynamicBrokersReader
test_class.var_name: dynamicBrokersReader
test_class.declarator: masterPath = "/brokers"
test_class.modifier: private
test_class.original_string: private String masterPath = "/brokers";
test_class.type: String
test_class.var_name: masterPath
test_class.declarator: topic = "testing"
test_class.modifier: private
test_class.original_string: private String topic = "testing";
test_class.type: String
test_class.var_name: topic
test_class.declarator: zookeeper
test_class.modifier: private
test_class.original_string: private CuratorFramework zookeeper;
test_class.type: CuratorFramework
test_class.var_name: zookeeper
test_class.declarator: server
test_class.modifier: private
test_class.original_string: private TestingServer server;
test_class.type: TestingServer
test_class.var_name: server
test_class.file: storm/storm-app/src/test/java/storm/kafka/DynamicBrokersReaderTest.java
test_class.identifier: DynamicBrokersReaderTest
test_case.body: @Test
      public void testGetBrokerInfo() throws Exception {
          String host = "localhost";
          int port = 9092;
          int partition = 0;
          addPartition(partition, host, port);
          GlobalPartitionInformation brokerInfo = dynamicBrokersReader.getBrokerInfo();
          assertEquals(1, brokerInfo.getOrderedPartitions().size());
          assertEquals(port, brokerInfo.getBrokerFor(partition).port);
          assertEquals(host, brokerInfo.getBrokerFor(partition).host);
      }
test_case.class_method_signature: DynamicBrokersReaderTest.testGetBrokerInfo()
test_case.full_signature: @Test public void testGetBrokerInfo()
test_case.identifier: testGetBrokerInfo
test_case.invocations: addPartition getBrokerInfo assertEquals size getOrderedPartitions assertEquals getBrokerFor assertEquals getBrokerFor
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testGetBrokerInfo()
focal_class.declarator: LOG = LoggerFactory.getLogger(DynamicBrokersReader.class)
focal_class.modifier: public static final
focal_class.original_string: public static final Logger LOG = LoggerFactory.getLogger(DynamicBrokersReader.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: _curator
focal_class.modifier: private
focal_class.original_string: private CuratorFramework _curator;
focal_class.type: CuratorFramework
focal_class.var_name: _curator
focal_class.declarator: _zkPath
focal_class.modifier: private
focal_class.original_string: private String _zkPath;
focal_class.type: String
focal_class.var_name: _zkPath
focal_class.declarator: _topic
focal_class.modifier: private
focal_class.original_string: private String _topic;
focal_class.type: String
focal_class.var_name: _topic
focal_class.file: storm/storm-app/src/main/java/storm/kafka/DynamicBrokersReader.java
focal_class.identifier: DynamicBrokersReader
focal_class.class_method_signature: DynamicBrokersReader.DynamicBrokersReader(Map conf, String zkStr, String zkPath, String topic)
focal_class.full_signature: public  DynamicBrokersReader(Map conf, String zkStr, String zkPath, String topic)
focal_class.identifier: DynamicBrokersReader
focal_class.modifiers: public
focal_class.parameters: (Map conf, String zkStr, String zkPath, String topic)
focal_class.signature:  DynamicBrokersReader(Map conf, String zkStr, String zkPath, String topic)
focal_class.class_method_signature: DynamicBrokersReader.getBrokerInfo()
focal_class.full_signature: public GlobalPartitionInformation getBrokerInfo()
focal_class.identifier: getBrokerInfo
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: GlobalPartitionInformation
focal_class.signature: GlobalPartitionInformation getBrokerInfo()
focal_class.class_method_signature: DynamicBrokersReader.getNumPartitions()
focal_class.full_signature: private int getNumPartitions()
focal_class.identifier: getNumPartitions
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int getNumPartitions()
focal_class.class_method_signature: DynamicBrokersReader.partitionPath()
focal_class.full_signature: public String partitionPath()
focal_class.identifier: partitionPath
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String partitionPath()
focal_class.class_method_signature: DynamicBrokersReader.brokerPath()
focal_class.full_signature: public String brokerPath()
focal_class.identifier: brokerPath
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String brokerPath()
focal_class.class_method_signature: DynamicBrokersReader.getLeaderFor(long partition)
focal_class.full_signature: private int getLeaderFor(long partition)
focal_class.identifier: getLeaderFor
focal_class.modifiers: private
focal_class.parameters: (long partition)
focal_class.return: int
focal_class.signature: int getLeaderFor(long partition)
focal_class.class_method_signature: DynamicBrokersReader.close()
focal_class.full_signature: public void close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: DynamicBrokersReader.getBrokerHost(byte[] contents)
focal_class.full_signature: private Broker getBrokerHost(byte[] contents)
focal_class.identifier: getBrokerHost
focal_class.modifiers: private
focal_class.parameters: (byte[] contents)
focal_class.return: Broker
focal_class.signature: Broker getBrokerHost(byte[] contents)
focal_method.body: 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;
      }
focal_method.class_method_signature: DynamicBrokersReader.getBrokerInfo()
focal_method.full_signature: public GlobalPartitionInformation getBrokerInfo()
focal_method.identifier: getBrokerInfo
focal_method.invocations: getNumPartitions brokerPath getLeaderFor forPath getData getBrokerHost addPartition error info
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: GlobalPartitionInformation
focal_method.signature: GlobalPartitionInformation getBrokerInfo()