test_class.declarator: reader
test_class.modifier: @Mock
      private
test_class.original_string: @Mock
      private DynamicBrokersReader reader;
test_class.type: DynamicBrokersReader
test_class.var_name: reader
test_class.declarator: dynamicPartitionConnections
test_class.modifier: @Mock
      private
test_class.original_string: @Mock
      private DynamicPartitionConnections dynamicPartitionConnections;
test_class.type: DynamicPartitionConnections
test_class.var_name: dynamicPartitionConnections
test_class.declarator: broker = new KafkaTestBroker()
test_class.modifier: private
test_class.original_string: private KafkaTestBroker broker = new KafkaTestBroker();
test_class.type: KafkaTestBroker
test_class.var_name: broker
test_class.declarator: server
test_class.modifier: private
test_class.original_string: private TestingServer server;
test_class.type: TestingServer
test_class.var_name: server
test_class.declarator: stormConf = new HashMap()
test_class.modifier: private
test_class.original_string: private Map stormConf = new HashMap();
test_class.type: Map
test_class.var_name: stormConf
test_class.declarator: spoutConfig
test_class.modifier: private
test_class.original_string: private SpoutConfig spoutConfig;
test_class.type: SpoutConfig
test_class.var_name: spoutConfig
test_class.declarator: state
test_class.modifier: private
test_class.original_string: private ZkState state;
test_class.type: ZkState
test_class.var_name: state
test_class.declarator: simpleConsumer
test_class.modifier: private
test_class.original_string: private SimpleConsumer simpleConsumer;
test_class.type: SimpleConsumer
test_class.var_name: simpleConsumer
test_class.file: storm/storm-app/src/test/java/storm/kafka/ZkCoordinatorTest.java
test_class.identifier: ZkCoordinatorTest
test_case.body: @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);
          }
      }
test_case.class_method_signature: ZkCoordinatorTest.testOnePartitionPerTask()
test_case.full_signature: @Test public void testOnePartitionPerTask()
test_case.identifier: testOnePartitionPerTask
test_case.invocations: buildCoordinators thenReturn when getBrokerInfo buildPartitionInfo getMyManagedPartitions assertEquals size assertEquals getPartition get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testOnePartitionPerTask()
focal_class.declarator: LOG = LoggerFactory.getLogger(ZkCoordinator.class)
focal_class.modifier: public static final
focal_class.original_string: public static final Logger LOG = LoggerFactory.getLogger(ZkCoordinator.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: _spoutConfig
focal_class.original_string: SpoutConfig _spoutConfig;
focal_class.type: SpoutConfig
focal_class.var_name: _spoutConfig
focal_class.declarator: _taskIndex
focal_class.original_string: int _taskIndex;
focal_class.type: int
focal_class.var_name: _taskIndex
focal_class.declarator: _totalTasks
focal_class.original_string: int _totalTasks;
focal_class.type: int
focal_class.var_name: _totalTasks
focal_class.declarator: _topologyInstanceId
focal_class.original_string: String _topologyInstanceId;
focal_class.type: String
focal_class.var_name: _topologyInstanceId
focal_class.declarator: _managers = new HashMap()
focal_class.original_string: Map<Partition, PartitionManager> _managers = new HashMap();
focal_class.type: Map<Partition, PartitionManager>
focal_class.var_name: _managers
focal_class.declarator: _cachedList
focal_class.original_string: List<PartitionManager> _cachedList;
focal_class.type: List<PartitionManager>
focal_class.var_name: _cachedList
focal_class.declarator: _lastRefreshTime = null
focal_class.original_string: Long _lastRefreshTime = null;
focal_class.type: Long
focal_class.var_name: _lastRefreshTime
focal_class.declarator: _refreshFreqMs
focal_class.original_string: int _refreshFreqMs;
focal_class.type: int
focal_class.var_name: _refreshFreqMs
focal_class.declarator: _connections
focal_class.original_string: DynamicPartitionConnections _connections;
focal_class.type: DynamicPartitionConnections
focal_class.var_name: _connections
focal_class.declarator: _reader
focal_class.original_string: DynamicBrokersReader _reader;
focal_class.type: DynamicBrokersReader
focal_class.var_name: _reader
focal_class.declarator: _state
focal_class.original_string: ZkState _state;
focal_class.type: ZkState
focal_class.var_name: _state
focal_class.declarator: _stormConf
focal_class.original_string: Map _stormConf;
focal_class.type: Map
focal_class.var_name: _stormConf
focal_class.file: storm/storm-app/src/main/java/storm/kafka/ZkCoordinator.java
focal_class.identifier: ZkCoordinator
focal_class.interfaces: implements PartitionCoordinator
focal_class.class_method_signature: ZkCoordinator.ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId)
focal_class.full_signature: public  ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId)
focal_class.identifier: ZkCoordinator
focal_class.modifiers: public
focal_class.parameters: (DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId)
focal_class.signature:  ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId)
focal_class.class_method_signature: ZkCoordinator.ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader)
focal_class.full_signature: public  ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader)
focal_class.identifier: ZkCoordinator
focal_class.modifiers: public
focal_class.parameters: (DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader)
focal_class.signature:  ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader)
focal_class.class_method_signature: ZkCoordinator.buildReader(Map stormConf, SpoutConfig spoutConfig)
focal_class.full_signature: private static DynamicBrokersReader buildReader(Map stormConf, SpoutConfig spoutConfig)
focal_class.identifier: buildReader
focal_class.modifiers: private static
focal_class.parameters: (Map stormConf, SpoutConfig spoutConfig)
focal_class.return: DynamicBrokersReader
focal_class.signature: DynamicBrokersReader buildReader(Map stormConf, SpoutConfig spoutConfig)
focal_class.class_method_signature: ZkCoordinator.getMyManagedPartitions()
focal_class.full_signature: @Override public List<PartitionManager> getMyManagedPartitions()
focal_class.identifier: getMyManagedPartitions
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: List<PartitionManager>
focal_class.signature: List<PartitionManager> getMyManagedPartitions()
focal_class.class_method_signature: ZkCoordinator.refresh()
focal_class.full_signature: @Override public void refresh()
focal_class.identifier: refresh
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void refresh()
focal_class.class_method_signature: ZkCoordinator.getManager(Partition partition)
focal_class.full_signature: @Override public PartitionManager getManager(Partition partition)
focal_class.identifier: getManager
focal_class.modifiers: @Override public
focal_class.parameters: (Partition partition)
focal_class.return: PartitionManager
focal_class.signature: PartitionManager getManager(Partition partition)
focal_method.body: @Override
      public List<PartitionManager> getMyManagedPartitions() {
          if (_lastRefreshTime == null || (System.currentTimeMillis() - _lastRefreshTime) > _refreshFreqMs) {
              refresh();
              _lastRefreshTime = System.currentTimeMillis();
          }
          return _cachedList;
      }
focal_method.class_method_signature: ZkCoordinator.getMyManagedPartitions()
focal_method.full_signature: @Override public List<PartitionManager> getMyManagedPartitions()
focal_method.identifier: getMyManagedPartitions
focal_method.invocations: currentTimeMillis refresh currentTimeMillis
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: List<PartitionManager>
focal_method.signature: List<PartitionManager> getMyManagedPartitions()