ZkCoordinator implements PartitionCoordinator { public synchronized boolean isZkOwnerExist(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath) && m_consumerId.equals(m_zk.readString(ownerPath))) { return true; } } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(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 isZkOwnerExist() { boolean rs = zkCoordinator.isZkOwnerExist(topic1, 2); assertFalse(rs); zkCoordinator.takePartition(topic1, 2); rs = zkCoordinator.isZkOwnerExist(topic1, 2); assertTrue(rs); }