ZkCoordinator implements PartitionCoordinator { @Override public List<PartitionManager> getMyManagedPartitions() { if (_lastRefreshTime == null || (System.currentTimeMillis() - _lastRefreshTime) > _refreshFreqMs) { refresh(); _lastRefreshTime = System.currentTimeMillis(); } return _cachedList; } ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId);  ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader); @Override List<PartitionManager> getMyManagedPartitions(); @Override void refresh(); @Override PartitionManager getManager(Partition partition); static final Logger LOG; }
@Test public void testOnePartitionPerTask() throws Exception { int totalTasks = 64; int partitionsPerTask = 1; List<ZkCoordinator> coordinatorList = buildCoordinators(totalTasks / partitionsPerTask); when(reader.getBrokerInfo()).thenReturn(TestUtils.buildPartitionInfo(totalTasks)); for (ZkCoordinator coordinator : coordinatorList) { List<PartitionManager> myManagedPartitions = coordinator.getMyManagedPartitions(); assertEquals(partitionsPerTask, myManagedPartitions.size()); assertEquals(coordinator._taskIndex, myManagedPartitions.get(0).getPartition().partition); } }