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; }
@Test public void testBrokerListDiscovery() throws Exception { List<String> brokers = KafkaUtils.retrieveBrokerListFromZK(curatorFramework); assertEquals(kafkaServersString.split(",").length, brokers.size()); assertEquals(servers.size(), brokers.size()); }
@Test public void testBrokerListDiscoveryWithNonExistingZKHost() throws Exception { CuratorFramework curatorFramework1 = Mockito.mock(CuratorFramework.class); CuratorZookeeperClient curatorZookeeperClient = Mockito.mock(CuratorZookeeperClient.class); ExistsBuilder existsBuilder = Mockito.mock(ExistsBuilder.class); Stat stat = Mockito.mock(Stat.class); GetChildrenBuilder childrenBuilder = Mockito.mock(GetChildrenBuilder.class); GetDataBuilder dataBuilder = Mockito.mock(GetDataBuilder.class); Mockito.when(curatorFramework1.getChildren()).thenReturn(childrenBuilder); Mockito.when(childrenBuilder.forPath(eq("/brokers/ids"))).thenReturn(Collections.<String>emptyList()); List<String> brokers = KafkaUtils.retrieveBrokerListFromZK(curatorFramework1); assertEquals(0, brokers.size()); }