DynamicCoordinator extends ZkCoordinator { @Override public synchronized Set<Integer> getToTakePartitions(String topic) { if (!m_allTopicPartitions.containsKey(topic)) return new HashSet<Integer>(); return getIdlePartitions(topic); } DynamicCoordinator(List<String> topics, String groupId,
			String consumerId, KafkaController.ZkConnector zk); @Override synchronized Map<String, Integer> getRebalanceCount(); @Override synchronized Set<Integer> getToTakePartitions(String topic); @Override synchronized Set<Integer> getToReleasePartitions(String topic,
			int count);  }
@Test public void testGetToTakePartitions() throws Exception { String groupId2 = "test_group2"; String topic2 = "Topic.test-2"; List<String> topics2 = new ArrayList<String>(); topics2.add(topic2); String consumerId2 = "D-00456-consumerId-2"; ZkCoordinator zkCoordinator2 = new DynamicCoordinator(topics2, groupId2, consumerId2, zkConnector); Set<Integer> toTake = zkCoordinator2.getToTakePartitions(topic2); assertTrue(toTake.isEmpty()); String path2 = zkCoordinator2.partitionPath(topic2); zkConnector.create(path2, false); zkConnector.create(path2 + "/0", false); zkConnector.create(path2 + "/1", false); toTake = zkCoordinator2.getToTakePartitions(topic2); assertEquals(2, toTake.size()); assertTrue(toTake.contains(0)); assertTrue(toTake.contains(1)); zkCoordinator2.takePartition(topic2, 0); toTake = zkCoordinator2.getToTakePartitions(topic2); assertEquals(1, toTake.size()); assertTrue(toTake.contains(1)); }