org.apache.storm.kafka

Class KafkaUtils

java.lang.Object

org.apache.storm.kafka.KafkaUtils

public final class KafkaUtils
extends java.lang.Object

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaUtils.KafkaOffsetMetric

Method Summary

Modifier and Type	Method and Description
static java.util.List<Partition>	calculatePartitionsForTask(java.util.List<GlobalPartitionInformation> partitons, int totalTasks, int taskIndex)
static ByteBufferMessageSet	fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) Fetch messages from kafka
static java.lang.Iterable<java.util.List<java.lang.Object>>	generateTuples(KafkaConfig kafkaConfig, Message msg, java.lang.String topic)
static java.lang.Iterable<java.util.List<java.lang.Object>>	generateTuples(MessageMetadataSchemeAsMultiScheme scheme, Message msg, Partition partition, long offset)
static long	getOffset(SimpleConsumer consumer, java.lang.String topic, int partition, KafkaConfig config)
static long	getOffset(SimpleConsumer consumer, java.lang.String topic, int partition, long startOffsetTime)
static IBrokerReader	makeBrokerReader(java.util.Map<java.lang.String,java.lang.Object> stormConf, KafkaConfig conf)
static java.lang.String	taskId(int taskIndex, int totalTasks)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

makeBrokerReader

public static IBrokerReader makeBrokerReader(java.util.Map<java.lang.String,java.lang.Object> stormConf,
                                             KafkaConfig conf)

getOffset

public static long getOffset(SimpleConsumer consumer,
                             java.lang.String topic,
                             int partition,
                             KafkaConfig config)

getOffset

public static long getOffset(SimpleConsumer consumer,
                             java.lang.String topic,
                             int partition,
                             long startOffsetTime)

fetchMessages

public static ByteBufferMessageSet fetchMessages(KafkaConfig config,
                                                 SimpleConsumer consumer,
                                                 Partition partition,
                                                 long offset)
                                          throws TopicOffsetOutOfRangeException,
                                                 FailedFetchException,
                                                 java.lang.RuntimeException

Fetch messages from kafka

Parameters:

config -

consumer -

partition -

offset -

Returns:

Throws:

TopicOffsetOutOfRangeException

FailedFetchException

java.lang.RuntimeException

generateTuples

public static java.lang.Iterable<java.util.List<java.lang.Object>> generateTuples(KafkaConfig kafkaConfig,
                                                                                  Message msg,
                                                                                  java.lang.String topic)

generateTuples

public static java.lang.Iterable<java.util.List<java.lang.Object>> generateTuples(MessageMetadataSchemeAsMultiScheme scheme,
                                                                                  Message msg,
                                                                                  Partition partition,
                                                                                  long offset)

calculatePartitionsForTask

public static java.util.List<Partition> calculatePartitionsForTask(java.util.List<GlobalPartitionInformation> partitons,
                                                                   int totalTasks,
                                                                   int taskIndex)

taskId

public static java.lang.String taskId(int taskIndex,
                                      int totalTasks)