ZkCoordinator implements PartitionCoordinator { @Override public synchronized boolean takePartition(String topic, int partition) { Set<Integer> partitionSet = m_myTopicPartitions.get(topic); if (partitionSet == null) return false; String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { return false; } if (m_zk.create(ownerPath, true)) { m_zk.writeString(ownerPath, m_consumerId); partitionSet.add(partition); return true; } else return false; } catch (RuntimeException e) { throw e; } catch (Exception e) { LOGGER.error( "Fail to take partition " + ownerPath, e); return false; } } 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 testTakePartition() { String ownerPath = zkCoordinator.ownerPath(topic1, 0); zkCoordinator.takePartition(topic1, 0); assertTrue(zkConnector.exists(ownerPath)); String owner = zkConnector.readString(ownerPath); assertEquals(consumerId, owner); }