test_class.declarator: broker
test_class.modifier: private
test_class.original_string: private KafkaTestBroker broker;
test_class.type: KafkaTestBroker
test_class.var_name: broker
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.declarator: config
test_class.modifier: private
test_class.original_string: private KafkaConfig config;
test_class.type: KafkaConfig
test_class.var_name: config
test_class.declarator: brokerHosts
test_class.modifier: private
test_class.original_string: private BrokerHosts brokerHosts;
test_class.type: BrokerHosts
test_class.var_name: brokerHosts
test_class.file: storm/storm-app/src/test/java/storm/kafka/KafkaUtilsTest.java
test_class.identifier: KafkaUtilsTest
test_case.body: @Test
      public void moreTasksThanPartitions() {
          GlobalPartitionInformation globalPartitionInformation = TestUtils.buildPartitionInfo(1);
          int numTasks = 2;
          assertEquals(1, KafkaUtils.calculatePartitionsForTask(globalPartitionInformation, numTasks, 0).size());
          assertEquals(0, KafkaUtils.calculatePartitionsForTask(globalPartitionInformation, numTasks, 1).size());
      }
test_case.class_method_signature: KafkaUtilsTest.moreTasksThanPartitions()
test_case.full_signature: @Test public void moreTasksThanPartitions()
test_case.identifier: moreTasksThanPartitions
test_case.invocations: buildPartitionInfo assertEquals size calculatePartitionsForTask assertEquals size calculatePartitionsForTask
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void moreTasksThanPartitions()
focal_class.declarator: LOG = LoggerFactory.getLogger(KafkaUtils.class)
focal_class.modifier: public static final
focal_class.original_string: public static final Logger LOG = LoggerFactory.getLogger(KafkaUtils.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: NO_OFFSET = -5
focal_class.modifier: private static final
focal_class.original_string: private static final int NO_OFFSET = -5;
focal_class.type: int
focal_class.var_name: NO_OFFSET
focal_class.file: storm/storm-app/src/main/java/storm/kafka/KafkaUtils.java
focal_class.identifier: KafkaUtils
focal_class.class_method_signature: KafkaUtils.makeBrokerReader(Map stormConf, KafkaConfig conf)
focal_class.full_signature: public static IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf)
focal_class.identifier: makeBrokerReader
focal_class.modifiers: public static
focal_class.parameters: (Map stormConf, KafkaConfig conf)
focal_class.return: IBrokerReader
focal_class.signature: IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf)
focal_class.class_method_signature: KafkaUtils.getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_class.full_signature: public static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_class.identifier: getOffset
focal_class.modifiers: public static
focal_class.parameters: (SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_class.return: long
focal_class.signature: long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_class.class_method_signature: KafkaUtils.getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime)
focal_class.full_signature: public static long getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime)
focal_class.identifier: getOffset
focal_class.modifiers: public static
focal_class.parameters: (SimpleConsumer consumer, String topic, int partition, long startOffsetTime)
focal_class.return: long
focal_class.signature: long getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime)
focal_class.class_method_signature: KafkaUtils.fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_class.full_signature: public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_class.identifier: fetchMessages
focal_class.modifiers: public static
focal_class.parameters: (KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_class.return: ByteBufferMessageSet
focal_class.signature: ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_class.class_method_signature: KafkaUtils.generateTuples(KafkaConfig kafkaConfig, Message msg)
focal_class.full_signature: public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg)
focal_class.identifier: generateTuples
focal_class.modifiers: public static
focal_class.parameters: (KafkaConfig kafkaConfig, Message msg)
focal_class.return: Iterable<List<Object>>
focal_class.signature: Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg)
focal_class.class_method_signature: KafkaUtils.calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_class.full_signature: public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_class.identifier: calculatePartitionsForTask
focal_class.modifiers: public static
focal_class.parameters: (GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_class.return: List<Partition>
focal_class.signature: List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_class.class_method_signature: KafkaUtils.logPartitionMapping(int totalTasks, int taskIndex, List<Partition> taskPartitions)
focal_class.full_signature: private static void logPartitionMapping(int totalTasks, int taskIndex, List<Partition> taskPartitions)
focal_class.identifier: logPartitionMapping
focal_class.modifiers: private static
focal_class.parameters: (int totalTasks, int taskIndex, List<Partition> taskPartitions)
focal_class.return: void
focal_class.signature: void logPartitionMapping(int totalTasks, int taskIndex, List<Partition> taskPartitions)
focal_class.class_method_signature: KafkaUtils.taskId(int taskIndex, int totalTasks)
focal_class.full_signature: public static String taskId(int taskIndex, int totalTasks)
focal_class.identifier: taskId
focal_class.modifiers: public static
focal_class.parameters: (int taskIndex, int totalTasks)
focal_class.return: String
focal_class.signature: String taskId(int taskIndex, int totalTasks)
focal_method.body: public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex) {
          Preconditions.checkArgument(taskIndex < totalTasks, "task index must be less that total tasks");
          List<Partition> partitions = partitionInformation.getOrderedPartitions();
          //kafka的日志分区数和spout的任务数.
          int numPartitions = partitions.size();
          //如果spout的任务数比日志分区数还要多, 说明有些task会空闲. 一般要一个任务处理多个日志分区, 即totalTasks <= partitions
          if (numPartitions < totalTasks) {
              LOG.warn("there are more tasks than partitions (tasks: " + totalTasks + "; partitions: " + numPartitions + "), some tasks will be idle");
          }
  
          List<Partition> taskPartitions = new ArrayList<Partition>();
          //假设有10个分区日志,有5个任务. taskIndex=0, 分配日志分区=0,5
          //taskIndex=1, 分配日志分区=1,6. 实际上是顺序为task分配一个日志分区: 每个task分配一个顺序的日志分区
          //如果还有剩余的日志分区没有指定task,再从头开始为所有的task继续顺序分配
          for (int i = taskIndex; i < numPartitions; i += totalTasks) {
              Partition taskPartition = partitions.get(i);
              taskPartitions.add(taskPartition);
          }
          logPartitionMapping(totalTasks, taskIndex, taskPartitions);
          return taskPartitions;
      }
focal_method.class_method_signature: KafkaUtils.calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_method.full_signature: public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_method.identifier: calculatePartitionsForTask
focal_method.invocations: checkArgument getOrderedPartitions size warn get add logPartitionMapping
focal_method.modifiers: public static
focal_method.parameters: (GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)
focal_method.return: List<Partition>
focal_method.signature: List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex)