StaticCoordinator extends ZkCoordinator { @Override public synchronized Set<Integer> getToTakePartitions(String topic) { Set<Integer> myPartitions = getMyPartitions(topic); Set<Integer> copy = new HashSet<Integer>(m_allocated); copy.removeAll(myPartitions); return copy; } StaticCoordinator(String config, 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"; String path2 = "/brokers/topics/" + topic2 + "/partitions"; zkConnector.create(path2, false); zkConnector.create(path2 + "/0", false); zkConnector.create(path2 + "/1", false); zkConnector.create(path2 + "/2", false); ZkCoordinator zkCoordinator2 = new StaticCoordinator("a,-1,1,2,5", topics2, groupId2, consumerId2, zkConnector); Set<Integer> toTake = zkCoordinator2.getToTakePartitions(topic2); assertEquals(2, toTake.size()); assertTrue(toTake.contains(1)); assertTrue(toTake.contains(2)); zkCoordinator2.takePartition(topic2, 1); toTake = zkCoordinator2.getToTakePartitions(topic2); assertEquals(1, toTake.size()); assertTrue(toTake.contains(2)); }