org.apache.crunch.kafka

Class KafkaUtils

java.lang.Object

org.apache.crunch.kafka.KafkaUtils

public class KafkaUtils
extends Object

Simple utilities for retrieving offset and Kafka information to assist in setting up and configuring a KafkaSource instance.

Field Summary

Fields

Modifier and Type	Field and Description
static String	KAFKA_EMPTY_RETRY_ATTEMPTS_KEY Configuration property for the number of retry attempts that will be made to Kafka in the event of getting empty responses.
static int	KAFKA_RETRY_ATTEMPTS_DEFAULT Default number of retry attempts.
static String	KAFKA_RETRY_ATTEMPTS_DEFAULT_STRING
static String	KAFKA_RETRY_ATTEMPTS_KEY Configuration property for the number of retry attempts that will be made to Kafka.
static int	KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT Default number of empty retry attempts.
static String	KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT_STRING

Constructor Summary

Constructors

Constructor and Description
KafkaUtils()

Method Summary

Modifier and Type	Method and Description
static org.apache.hadoop.conf.Configuration	addKafkaConnectionProperties(Properties properties, org.apache.hadoop.conf.Configuration config) Adds the properties to the provided config instance.
static Map<org.apache.kafka.common.TopicPartition,Long>	getBrokerOffsets(Properties properties, long time, String... topics) Retrieves the offset values for an array of topics at the specified time.
static Properties	getKafkaConnectionProperties(org.apache.hadoop.conf.Configuration config) Converts the provided config into a Properties object to connect with Kafka.

Methods inherited from class java.lang.Object

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

Field Detail

KAFKA_RETRY_ATTEMPTS_KEY

public static final String KAFKA_RETRY_ATTEMPTS_KEY

Configuration property for the number of retry attempts that will be made to Kafka.

See Also:

Constant Field Values

KAFKA_RETRY_ATTEMPTS_DEFAULT

public static final int KAFKA_RETRY_ATTEMPTS_DEFAULT

Default number of retry attempts.

See Also:

Constant Field Values

KAFKA_RETRY_ATTEMPTS_DEFAULT_STRING

public static final String KAFKA_RETRY_ATTEMPTS_DEFAULT_STRING

KAFKA_EMPTY_RETRY_ATTEMPTS_KEY

public static final String KAFKA_EMPTY_RETRY_ATTEMPTS_KEY

Configuration property for the number of retry attempts that will be made to Kafka in the event of getting empty responses.

See Also:

Constant Field Values

KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT

public static final int KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT

Default number of empty retry attempts.

See Also:

Constant Field Values

KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT_STRING

public static final String KAFKA_RETRY_EMPTY_ATTEMPTS_DEFAULT_STRING

Constructor Detail

KafkaUtils

public KafkaUtils()

Method Detail

getKafkaConnectionProperties

public static Properties getKafkaConnectionProperties(org.apache.hadoop.conf.Configuration config)

Converts the provided config into a Properties object to connect with Kafka.

Parameters:

config - the config to read properties

Returns:

a properties instance populated with all of the values inside the provided config.

addKafkaConnectionProperties

public static org.apache.hadoop.conf.Configuration addKafkaConnectionProperties(Properties properties,
                                                                                org.apache.hadoop.conf.Configuration config)

Adds the properties to the provided config instance.

Parameters:

properties - the properties to add to the config.

config - the configuration instance to be modified.

Returns:

the config instance with the populated properties

getBrokerOffsets

public static Map<org.apache.kafka.common.TopicPartition,Long> getBrokerOffsets(Properties properties,
                                                                                long time,
                                                                                String... topics)

Retrieves the offset values for an array of topics at the specified time.

If the Kafka cluster does not have the logs for the partition at the specified time or if the topic did not exist at that time this will instead return the earliest offset for that partition.

Parameters:

properties - the properties containing the configuration for kafka

time - the time at which we want to know what the offset values were

topics - the topics we want to know the offset values of

Returns:

the offset values for an array of topics at the specified time

Throws:

IllegalArgumentException - if properties is null or if topics is null or empty or if any of the topics are null, empty or blank, or if there is an error parsing the properties.

IllegalStateException - if there is an error communicating with the Kafka cluster to retrieve information.