public String getBrokerInfo() {
    	String brokerStr = "";
        try {
			int numPartitionsForTopic = getNumPartitions();
			String brokerInfoPath = brokerPath();
			for (int partition = 0; partition < numPartitionsForTopic; partition++) {
				int leader = getLeaderFor(partition);
				String path = brokerInfoPath + "/" + leader;
				try {
					byte[] hostPortData = _curator.getData().forPath(path);
					brokerStr = brokerStr + getBrokerHost(hostPortData);
					if(partition!=numPartitionsForTopic-1)
						brokerStr +=",";
				} catch(org.apache.zookeeper.KeeperException.NoNodeException e) {
					LOG.error("Node {} does not exist ", path);
				}
			}
			_curator.close();
        } catch(Exception e) {
            throw new RuntimeException(e);
        }
		LOG.info("Read partition info from zookeeper: " + brokerStr);
        return brokerStr;
    }