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 generateTuplesWithKeyAndKeyValueScheme() {
          config.scheme = new KeyValueSchemeAsMultiScheme(new StringKeyValueScheme());
          config.useStartOffsetTimeIfOffsetOutOfRange = false;
          String value = "value";
          String key = "key";
          createTopicAndSendMessage(key, value);
          ByteBufferMessageSet messageAndOffsets = getLastMessage();
          for (MessageAndOffset msg : messageAndOffsets) {
              Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message());
              assertEquals(ImmutableMap.of(key, value), lists.iterator().next().get(0));
          }
      }
test_case.class_method_signature: KafkaUtilsTest.generateTuplesWithKeyAndKeyValueScheme()
test_case.full_signature: @Test public void generateTuplesWithKeyAndKeyValueScheme()
test_case.identifier: generateTuplesWithKeyAndKeyValueScheme
test_case.invocations: createTopicAndSendMessage getLastMessage generateTuples message assertEquals of get next iterator
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void generateTuplesWithKeyAndKeyValueScheme()
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 Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg) {
          Iterable<List<Object>> tups;
          //消息的负载是字节缓冲区, 转为序列化后的字节数组
          ByteBuffer payload = msg.payload();
          if (payload == null) {
              return null;
          }
          ByteBuffer key = msg.key();
          if (key != null && kafkaConfig.scheme instanceof KeyValueSchemeAsMultiScheme) {
              tups = ((KeyValueSchemeAsMultiScheme) kafkaConfig.scheme).deserializeKeyAndValue(Utils.toByteArray(key), Utils.toByteArray(payload));
          } else {
              tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload));
          }
          return tups;
      }
focal_method.class_method_signature: KafkaUtils.generateTuples(KafkaConfig kafkaConfig, Message msg)
focal_method.full_signature: public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg)
focal_method.identifier: generateTuples
focal_method.invocations: payload key deserializeKeyAndValue toByteArray toByteArray deserialize toByteArray
focal_method.modifiers: public static
focal_method.parameters: (KafkaConfig kafkaConfig, Message msg)
focal_method.return: Iterable<List<Object>>
focal_method.signature: Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg)