StaticCoordinator extends ZkCoordinator { @Override public synchronized Set<Integer> getToReleasePartitions(String topic, int count) { Set<Integer> myPartitions = getMyPartitions(topic); Set<Integer> copy = new HashSet<Integer>(myPartitions); copy.removeAll(m_allocated); 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 testGetToReleasePartitions() throws Exception { String groupId3 = "test_group3"; String topic3 = "Topic.test-3"; List<String> topics3 = new ArrayList<String>(); topics3.add(topic3); String consumerId3 = "D-00456-consumerId-3"; String path3 = "/brokers/topics/" + topic3 + "/partitions"; zkConnector.create(path3, false); zkConnector.create(path3 + "/0", false); zkConnector.create(path3 + "/1", false); zkConnector.create(path3 + "/2", false); zkConnector.create(path3 + "/3", false); ZkCoordinator zkCoordinator3 = new StaticCoordinator("a,-1,1,2,5", topics3, groupId3, consumerId3, zkConnector); zkCoordinator3.takePartition(topic3, 0); zkCoordinator3.takePartition(topic3, 1); zkCoordinator3.takePartition(topic3, 2); zkCoordinator3.takePartition(topic3, 3); Set<Integer> toRelease = zkCoordinator3.getToReleasePartitions(topic3, 2); assertEquals(2, toRelease.size()); Set<Integer> partitions = new HashSet<Integer>(); partitions.add(0); partitions.add(3); assertTrue(partitions.containsAll(toRelease)); }