ZkCoordinator implements PartitionCoordinator { protected synchronized Set<Integer> getRandomReleasePartitions( String topic, int count) { try { Set<Integer> toRelease = new HashSet<Integer>(); Set<Integer> myPartitions = m_myTopicPartitions.get(topic); if (myPartitions == null || myPartitions.size() < count) return toRelease; Iterator<Integer> it = myPartitions.iterator(); while (it.hasNext() && toRelease.size() < count) { toRelease.add(it.next()); } return toRelease; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } 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 testGetRandomReleasePartitions() { zkCoordinator.takePartition(topic2, 0); zkCoordinator.takePartition(topic2, 1); zkCoordinator.takePartition(topic2, 2); zkCoordinator.getMyPartitions(topic2); Set<Integer> toRelease = zkCoordinator.getRandomReleasePartitions( topic2, 2); Set<Integer> partitions = new HashSet<Integer>(); partitions.add(0); partitions.add(1); partitions.add(2); assertEquals(2, toRelease.size()); assertTrue(partitions.containsAll(toRelease)); }