ZkCoordinator implements PartitionCoordinator { @Override public synchronized boolean isNewPartition(String topic, int partition) { StringBuffer sb = new StringBuffer(m_offsetPath); sb.append("/").append(topic).append("/").append(partition); boolean isOffsetExist = m_zk.exists(sb.toString()); boolean newAdd = false; Set<Integer> partitions = m_newTopicPartitions.get(topic); if (partitions != null && partitions.contains(partition)) newAdd = true; return newAdd && !isOffsetExist; } 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 testIsNewPartition() throws Exception { String topic11 = "Topic.test-11"; String path11 = "/brokers/topics/" + topic11 + "/partitions"; zkConnector.create(path11, false); zkConnector.create(path11 + "/0", false); String topic12 = "Topic.test-12"; String path12 = "/brokers/topics/" + topic12 + "/partitions"; zkConnector.create(path12, false); zkConnector.create(path12 + "/0", false); List<String> topics = new ArrayList<String>(); topics.add(topic11); topics.add(topic12); ZkCoordinator zkCoordinator = new DynamicCoordinator(topics, groupId, consumerId, zkConnector); zkCoordinator.getAllPartitions(topic11); zkCoordinator.getAllPartitions(topic12); zkConnector.create(path11 + "/1", false); zkConnector.create(path12 + "/1", false); zkCoordinator.getAllPartitions(topic11); zkCoordinator.getAllPartitions(topic12); assertTrue(zkCoordinator.isNewPartition(topic11, 1)); assertTrue(zkCoordinator.isNewPartition(topic12, 1)); zkConnector.create(path11 + "/2", false); zkConnector.create(path12 + "/2", false); zkCoordinator.getAllPartitions(topic11); zkCoordinator.getAllPartitions(topic12); assertTrue(zkCoordinator.isNewPartition(topic11, 2)); assertTrue(zkCoordinator.isNewPartition(topic12, 2)); }