@Test public void testBrokerListDiscovery() throws Exception { List<String> brokers = KafkaUtils.retrieveBrokerListFromZK(curatorFramework); assertEquals(kafkaServersString.split(",").length, brokers.size()); assertEquals(servers.size(), brokers.size()); }
public static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework) throws Exception { final List<String> brokers = new LinkedList<>(); List<String> znodes = curatorFramework.getChildren().forPath("/brokers/ids"); for (String znode : znodes) { Map<String, Object> stringObjectMap = TrumpetHelper.toMap(curatorFramework.getData().forPath("/brokers/ids/" + znode)); brokers.add(stringObjectMap.get("host").toString() + ":" + stringObjectMap.get("port")); } return brokers; }
KafkaUtils { public static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework) throws Exception { final List<String> brokers = new LinkedList<>(); List<String> znodes = curatorFramework.getChildren().forPath("/brokers/ids"); for (String znode : znodes) { Map<String, Object> stringObjectMap = TrumpetHelper.toMap(curatorFramework.getData().forPath("/brokers/ids/" + znode)); brokers.add(stringObjectMap.get("host").toString() + ":" + stringObjectMap.get("port")); } return brokers; } }
KafkaUtils { public static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework) throws Exception { final List<String> brokers = new LinkedList<>(); List<String> znodes = curatorFramework.getChildren().forPath("/brokers/ids"); for (String znode : znodes) { Map<String, Object> stringObjectMap = TrumpetHelper.toMap(curatorFramework.getData().forPath("/brokers/ids/" + znode)); brokers.add(stringObjectMap.get("host").toString() + ":" + stringObjectMap.get("port")); } return brokers; }  }
KafkaUtils { public static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework) throws Exception { final List<String> brokers = new LinkedList<>(); List<String> znodes = curatorFramework.getChildren().forPath("/brokers/ids"); for (String znode : znodes) { Map<String, Object> stringObjectMap = TrumpetHelper.toMap(curatorFramework.getData().forPath("/brokers/ids/" + znode)); brokers.add(stringObjectMap.get("host").toString() + ":" + stringObjectMap.get("port")); } return brokers; }  static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework); static void createTopic(String topic, CuratorFramework curatorFramework); static void createTopic(String topic, int partitions, int replication, CuratorFramework curatorFramework); }
KafkaUtils { public static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework) throws Exception { final List<String> brokers = new LinkedList<>(); List<String> znodes = curatorFramework.getChildren().forPath("/brokers/ids"); for (String znode : znodes) { Map<String, Object> stringObjectMap = TrumpetHelper.toMap(curatorFramework.getData().forPath("/brokers/ids/" + znode)); brokers.add(stringObjectMap.get("host").toString() + ":" + stringObjectMap.get("port")); } return brokers; }  static List<String> retrieveBrokerListFromZK(final CuratorFramework curatorFramework); static void createTopic(String topic, CuratorFramework curatorFramework); static void createTopic(String topic, int partitions, int replication, CuratorFramework curatorFramework); static final int DEFAULT_NUM_OF_PARTITION; static final int DEFAULT_REPLICATION; }