ZkCoordinator implements PartitionCoordinator { protected synchronized Set<Integer> getAllPartitions(String topic) { Set<Integer> partitions = new HashSet<Integer>(); try { String path = partitionPath(topic); if (!m_zk.exists(path)) { LOGGER.warn( "Node " + path + " does not exist in zookeeper."); return partitions; } List<String> pstrs = m_zk.getChildren(path); if (pstrs == null || pstrs.isEmpty()) { LOGGER.warn( "No partitions available for topic " + topic); return partitions; } for (String pstr : pstrs) { int pid = Integer.parseInt(pstr); partitions.add(pid); } } catch (Exception e) { LOGGER.error( "Error occurs in getting partitions of " + topic, e); } findNewlyAddPartitions(topic, partitions); m_allTopicPartitions.put(topic, new HashSet<Integer>(partitions)); return partitions; } ZkCoordinator(List<String> topics, String groupId,
			String consumerId, KafkaController.ZkConnector zk); @Override synchronized void registerConsumer(); @Override synchronized void unregisterConsumer(); synchronized void reRegisterConsumer(boolean force); @Override synchronized boolean isNewPartition(String topic, int partition); @Override synchronized boolean takePartition(String topic, int partition); synchronized boolean resetOwner(String topic, int partition); @Override synchronized void releasePartition(String topic, int partition); @Override synchronized Set<Integer> getMyPartitions(String topic); synchronized boolean isZkOwnerExist(String topic, int partition);  }
@Test public void testGetAllPartitions() { Set<Integer> partitions = zkCoordinator.getAllPartitions(topic1); assertTrue(partitions.isEmpty()); String path = zkCoordinator.partitionPath(topic1); zkConnector.create(path, false); partitions = zkCoordinator.getAllPartitions(topic1); assertTrue(partitions.isEmpty()); zkConnector.create(path + "/6", false); zkConnector.create(path + "/7", false); zkConnector.create(path + "/8", false); partitions = zkCoordinator.getAllPartitions(topic1); assertFalse(partitions.isEmpty()); assertTrue(partitions.contains(6)); assertTrue(partitions.contains(7)); assertTrue(partitions.contains(8)); }