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 getOffsetFromConfigAndFroceFromStart() {
          config.forceFromStart = true;
          config.startOffsetTime = OffsetRequest.EarliestTime();
          createTopicAndSendMessage();
          long earliestOffset = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.EarliestTime());
          long offsetFromConfig = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, config);
          assertThat(earliestOffset, is(equalTo(offsetFromConfig)));
      }
test_case.class_method_signature: KafkaUtilsTest.getOffsetFromConfigAndFroceFromStart()
test_case.full_signature: @Test public void getOffsetFromConfigAndFroceFromStart()
test_case.identifier: getOffsetFromConfigAndFroceFromStart
test_case.invocations: EarliestTime createTopicAndSendMessage getOffset EarliestTime getOffset assertThat is equalTo
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void getOffsetFromConfigAndFroceFromStart()
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 long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config) {
          long startOffsetTime = kafka.api.OffsetRequest.LatestTime();
          if ( config.forceFromStart ) {
              startOffsetTime = config.startOffsetTime;
          }
          return getOffset(consumer, topic, partition, startOffsetTime);
      }
focal_method.class_method_signature: KafkaUtils.getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_method.full_signature: public static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_method.identifier: getOffset
focal_method.invocations: LatestTime getOffset
focal_method.modifiers: public static
focal_method.parameters: (SimpleConsumer consumer, String topic, int partition, KafkaConfig config)
focal_method.return: long
focal_method.signature: long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config)