KafkaUtils { 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; }  static IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf); static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config); static long getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime); static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset); static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg); static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex); static String taskId(int taskIndex, int totalTasks); static final Logger LOG; }
@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 public void generateTuplesWithValueSchemeAndKeyValueMessage() { config.scheme = new SchemeAsMultiScheme(new StringScheme()); 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(value, lists.iterator().next().get(0)); } }