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(expected = FailedFetchException.class)
      public void fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled() throws Exception {
          config.useStartOffsetTimeIfOffsetOutOfRange = false;
          KafkaUtils.fetchMessages(config, simpleConsumer,
                  new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), -99);
      }
test_case.class_method_signature: KafkaUtilsTest.fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled()
test_case.full_signature: @Test(expected = FailedFetchException.class) public void fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled()
test_case.identifier: fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled
test_case.invocations: fetchMessages fromString getBrokerConnectionString
test_case.modifiers: @Test(expected = FailedFetchException.class) public
test_case.parameters: ()
test_case.return: void
test_case.signature: void fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled()
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 ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
              throws TopicOffsetOutOfRangeException, FailedFetchException,RuntimeException {
          ByteBufferMessageSet msgs = null;
          String topic = config.topic;
          int partitionId = partition.partition;
  
          FetchRequestBuilder builder = new FetchRequestBuilder();
          //Builder链式调用. fetchSizeBytes表示一次要获取多少数据量. 所以从offset开始,一次获取不止一条消息
          FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes).
                  clientId(config.clientId).maxWait(config.fetchMaxWait).build();
          FetchResponse fetchResponse;
          try {
              //消费者传入获取请求,得到获取响应
              fetchResponse = consumer.fetch(fetchRequest);
          } catch (Exception e) {
              if (e instanceof ConnectException ||
                      e instanceof SocketTimeoutException ||
                      e instanceof IOException ||
                      e instanceof UnresolvedAddressException
                      ) {
                  LOG.warn("Network error when fetching messages:", e);
                  throw new FailedFetchException(e);
              } else {
                  throw new RuntimeException(e);
              }
          }
          // 主要处理offset outofrange的case，通过getOffset从earliest或latest读
          if (fetchResponse.hasError()) {
              KafkaError error = KafkaError.getError(fetchResponse.errorCode(topic, partitionId));
              if (error.equals(KafkaError.OFFSET_OUT_OF_RANGE) && config.useStartOffsetTimeIfOffsetOutOfRange) {
                  String msg = "Got fetch request with offset out of range: [" + offset + "]";
                  LOG.warn(msg);
                  throw new TopicOffsetOutOfRangeException(msg);
              } else {
                  String message = "Error fetching data from [" + partition + "] for topic [" + topic + "]: [" + error + "]";
                  LOG.error(message);
                  throw new FailedFetchException(message);
              }
          } else {
              msgs = fetchResponse.messageSet(topic, partitionId);
          }
          return msgs;
      }
focal_method.class_method_signature: KafkaUtils.fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_method.full_signature: public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_method.identifier: fetchMessages
focal_method.invocations: build maxWait clientId addFetch fetch warn hasError getError errorCode equals warn error messageSet
focal_method.modifiers: public static
focal_method.parameters: (KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)
focal_method.return: ByteBufferMessageSet
focal_method.signature: ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset)