ZkCoordinator implements PartitionCoordinator { @Override public synchronized void releasePartition(String topic, int partition) { Set<Integer> partitionSet = m_myTopicPartitions.get(topic); if (partitionSet == null) return; String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath) && m_consumerId.equals(m_zk.readString(ownerPath))) { m_zk.delete(ownerPath); } partitionSet.remove(partition); } catch (Exception e) { LOGGER.error( m_consumerId + "Fail to release partition " + ownerPath, 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 testReleasePartition() { String ownerPath = zkCoordinator.ownerPath(topic1, 3); zkCoordinator.takePartition(topic1, 3); assertTrue(zkConnector.exists(ownerPath)); zkCoordinator.releasePartition(topic1, 3); assertFalse(zkConnector.exists(ownerPath)); }