org.apache.storm.kafka

Class KafkaUtils

java.lang.Object

org.apache.storm.kafka.KafkaUtils

public class KafkaUtils
extends Object

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaUtils.KafkaOffsetMetric

Constructor Summary

Constructors

Constructor and Description
KafkaUtils()

Method Summary

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

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaUtils

public KafkaUtils()

Method Detail

makeBrokerReader

public static IBrokerReader makeBrokerReader(Map stormConf,
                                             KafkaConfig conf)

getOffset

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

getOffset

public static long getOffset(kafka.javaapi.consumer.SimpleConsumer consumer,
                             String topic,
                             int partition,
                             long startOffsetTime)

fetchMessages

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

Throws:

TopicOffsetOutOfRangeException

FailedFetchException

RuntimeException

generateTuples

public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig,
                                                    kafka.message.Message msg,
                                                    String topic)

generateTuples

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

calculatePartitionsForTask

public static List<Partition> calculatePartitionsForTask(List<GlobalPartitionInformation> partitons,
                                                         int totalTasks,
                                                         int taskIndex,
                                                         int taskId)

taskPrefix

public static String taskPrefix(int taskIndex,
                                int totalTasks,
                                int taskId)