ZkCoordinator implements PartitionCoordinator { @Override public synchronized Set<Integer> getMyPartitions(String topic) { Set<Integer> partitions = new HashSet<Integer>(); try { String ownerPath = ownerPath(topic); if (!m_zk.exists(ownerPath)) return partitions; List<String> pstrs = m_zk.getChildren(ownerPath); if (pstrs == null || pstrs.isEmpty()) return partitions; for (String pstr : pstrs) { int pid = Integer.parseInt(pstr); String path = ownerPath(topic, pid); if (!m_zk.exists(path)) continue; String ownerId = m_zk.readString(path); if (ownerId != null && ownerId.equals(m_consumerId)) { partitions.add(pid); } } } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } m_myTopicPartitions.put(topic, new HashSet<Integer>(partitions)); return partitions; } 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 testGetMyPartitions() { zkCoordinator.takePartition(topic1, 4); zkCoordinator.takePartition(topic1, 5); Set<Integer> partitions = zkCoordinator.getMyPartitions(topic1); assertTrue(partitions.contains(4)); assertTrue(partitions.contains(5)); }