docs: private Map<String, SortedSet<Partition>> getZKTopicPartitionInfo(Map<Integer, Broker> allBrokers) {
          final Map<String, SortedSet<Partition>> brokerPartitionsPerTopic = new HashMap<String, SortedSet<Partition>>();
          ZkUtils.makeSurePersistentPathExists(zkClient, ZkUtils.BrokerTopicsPath);
          List<String> topics = ZkUtils.getChildrenParentMayNotExist(zkClient, ZkUtils.BrokerTopicsPath);
          for (String topic : topics) {
              // find the number of broker partitions registered for this topic
              String brokerTopicPath = ZkUtils.BrokerTopicsPath + "/" + topic;
              List<String> brokerList = ZkUtils.getChildrenParentMayNotExist(zkClient, brokerTopicPath);
              //
              final SortedSet<Partition> sortedBrokerPartitions = new TreeSet<Partition>();
              final Set<Integer> existBids = new HashSet<Integer>();
              for (String bid : brokerList) {
                  final int ibid = Integer.parseInt(bid);
                  final String numPath = brokerTopicPath + "/" + bid;
                  final Integer numPartition = Integer.valueOf(ZkUtils.readData(zkClient, numPath));
                  for (int i = 0; i < numPartition.intValue(); i++) {
                      sortedBrokerPartitions.add(new Partition(ibid, i));
                  }
                  existBids.add(ibid);
              }
              // add all brokers after topic created
              for(Integer bid:allBrokers.keySet()){
                  if(!existBids.contains(bid)){
                      sortedBrokerPartitions.add(new Partition(bid,0));// this broker run after topic created
                  }
              }
              logger.debug("Broker ids and # of partitions on each for topic: " + topic + " = " + sortedBrokerPartitions);
              brokerPartitionsPerTopic.put(topic, sortedBrokerPartitions);
          }
          return brokerPartitionsPerTopic;
      }