ZkCoordinator implements PartitionCoordinator { protected synchronized List<String> getAllConsumers() { return m_zk.getChildren(m_consumerPath); } 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 testGetAllConsumers() throws Exception { String consumerId1 = "consumerId1"; String consumerId2 = "consumerId2"; DynamicCoordinator consumer1 = new DynamicCoordinator(topics, groupId, consumerId1, zkConnector); consumer1.registerConsumer(); DynamicCoordinator consumer2 = new DynamicCoordinator(topics, groupId, consumerId2, zkConnector); consumer2.registerConsumer(); zkCoordinator.registerConsumer(); List<String> consumers = zkCoordinator.getAllConsumers(); assertTrue(consumers.contains(consumerId1)); assertTrue(consumers.contains(consumerId2)); assertTrue(consumers.contains(consumerId)); }