com.github.charithe.kafka

Class KafkaJunitRule

java.lang.Object

org.junit.rules.ExternalResource

com.github.charithe.kafka.KafkaJunitRule

All Implemented Interfaces:

org.junit.rules.TestRule

public class KafkaJunitRule
extends org.junit.rules.ExternalResource

Constructor Summary

Constructors

Constructor and Description
KafkaJunitRule()
KafkaJunitRule(int kafkaPort)
KafkaJunitRule(int kafkaPort, int zookeeperPort)
KafkaJunitRule(int kafkaPort, int zookeeperPort, Properties brokerProperties)

Method Summary

Modifier and Type	Method and Description
protected void	after()
protected void	before()
Properties	consumerConfig() Create a consumer configuration.
Properties	consumerConfig(boolean enableAutoCommit) Create a consumer configuration.
<K,V> org.apache.kafka.clients.consumer.KafkaConsumer<K,V>	createConsumer(org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer, org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer) Create a Kafka consumer using consumerConfig()
<K,V> org.apache.kafka.clients.consumer.KafkaConsumer<K,V>	createConsumer(org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer, org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer, boolean autoCommitEnabled) Create a Kafka consumer using consumerConfig(boolean)
<K,V> org.apache.kafka.clients.producer.KafkaProducer<K,V>	createProducer(org.apache.kafka.common.serialization.Serializer<K> keySerializer, org.apache.kafka.common.serialization.Serializer<V> valueSerializer) Create a Kafka producer using producerConfig()
org.apache.kafka.clients.consumer.KafkaConsumer<String,String>	createStringConsumer() Create a Kafka consumer that reads messages with String key and values.
org.apache.kafka.clients.consumer.KafkaConsumer<String,String>	createStringConsumer(boolean autoCommitEnabled) Create a Kafka consumer that reads messages with String key and values.
org.apache.kafka.clients.producer.KafkaProducer<String,String>	createStringProducer() Create a Kafka producer that produces messages with String keys and values
int	kafkaBrokerPort() Get the kafka broker port
Path	kafkaLogDir() Get the Kafka log directory
<K,V> com.google.common.util.concurrent.ListenableFuture<List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>>>	pollMessages(String topic, int numMessagesToPoll, org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer, org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer) Poll the specified topic for messages
com.google.common.util.concurrent.ListenableFuture<List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>>>	pollStringMessages(String topic, int numMessagesToPoll) Poll the specified topic for String key-valued messages
Properties	producerConfig() Create a producer configuration.
<T> List<T>	readMessages(org.apache.kafka.clients.consumer.KafkaConsumer<T,T> consumer, String topic, int expectedMessages, int timeoutSeconds) Deprecated. Use pollMessages(String, int, Deserializer, Deserializer) instead. This method is deprecated due to the switch to Kafka 0.9. The new API can return more than the expected number of messages unless manual offset management is performed. Since it is impossible to ensure that the passed-in KafkaConsumer has auto commit disabled, this method should not be used.
List<String>	readStringMessages(String topic, int expectedMessages, int timeoutSeconds) Deprecated. Use pollStringMessages(String, int) instead This method is deprecated because it silently creates a new consumer on each invocation -- which could lead to some mesages being read more than once. It also suffers from the problems described in the deprecation notice for readMessages(KafkaConsumer, String, int, int) as well.
void	shutdownKafka() Shutdown Kafka Broker before the test termination to test consumer exceptions
void	startKafka() Starts the server
String	zookeeperConnectionString() Get the zookeeper connection string
int	zookeeperPort() Get the zookeeper port

Methods inherited from class org.junit.rules.ExternalResource

apply

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaJunitRule

public KafkaJunitRule()

KafkaJunitRule

public KafkaJunitRule(int kafkaPort)

KafkaJunitRule

public KafkaJunitRule(int kafkaPort,
                      int zookeeperPort)

KafkaJunitRule

public KafkaJunitRule(int kafkaPort,
                      int zookeeperPort,
                      Properties brokerProperties)

Method Detail

before

protected void before()
               throws Throwable

Overrides:

before in class org.junit.rules.ExternalResource

Throws:

Throwable

after

protected void after()

Overrides:

after in class org.junit.rules.ExternalResource

shutdownKafka

public void shutdownKafka()

Shutdown Kafka Broker before the test termination to test consumer exceptions

startKafka

public void startKafka()

Starts the server

producerConfig

public Properties producerConfig()

Create a producer configuration.

consumerConfig

public Properties consumerConfig()

Create a consumer configuration. Auto commit is enabled.

Returns:

consumerConfig

public Properties consumerConfig(boolean enableAutoCommit)

Create a consumer configuration. Offset is set to "earliest".

createStringConsumer

public org.apache.kafka.clients.consumer.KafkaConsumer<String,String> createStringConsumer()

Create a Kafka consumer that reads messages with String key and values. Auto commit is enabled for the consumer.

Returns:

KafkaConsumer

createStringConsumer

public org.apache.kafka.clients.consumer.KafkaConsumer<String,String> createStringConsumer(boolean autoCommitEnabled)

Create a Kafka consumer that reads messages with String key and values.

Parameters:

autoCommitEnabled - Set to true to enable auto commit

Returns:

KafkaConsumer

createConsumer

public <K,V> org.apache.kafka.clients.consumer.KafkaConsumer<K,V> createConsumer(org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                                                 org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer)

Create a Kafka consumer using consumerConfig()

Returns:

KafkaConsumer

createConsumer

public <K,V> org.apache.kafka.clients.consumer.KafkaConsumer<K,V> createConsumer(org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                                                 org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer,
                                                                                 boolean autoCommitEnabled)

Create a Kafka consumer using consumerConfig(boolean)

Returns:

KafkaConsumer

createStringProducer

public org.apache.kafka.clients.producer.KafkaProducer<String,String> createStringProducer()

Create a Kafka producer that produces messages with String keys and values

createProducer

public <K,V> org.apache.kafka.clients.producer.KafkaProducer<K,V> createProducer(org.apache.kafka.common.serialization.Serializer<K> keySerializer,
                                                                                 org.apache.kafka.common.serialization.Serializer<V> valueSerializer)

Create a Kafka producer using producerConfig()

Returns:

KafkaProducer

readStringMessages

@Deprecated
public List<String> readStringMessages(String topic,
                                                   int expectedMessages,
                                                   int timeoutSeconds)
                                            throws TimeoutException

Deprecated. Use pollStringMessages(String, int) instead

This method is deprecated because it silently creates a new consumer on each invocation -- which could lead to some mesages being read more than once. It also suffers from the problems described in the deprecation notice for readMessages(KafkaConsumer, String, int, int) as well.

Throws:

TimeoutException

readMessages

@Deprecated
public <T> List<T> readMessages(org.apache.kafka.clients.consumer.KafkaConsumer<T,T> consumer,
                                            String topic,
                                            int expectedMessages,
                                            int timeoutSeconds)
                                     throws TimeoutException

Deprecated. Use pollMessages(String, int, Deserializer, Deserializer) instead.

This method is deprecated due to the switch to Kafka 0.9. The new API can return more than the expected number of messages unless manual offset management is performed. Since it is impossible to ensure that the passed-in KafkaConsumer has auto commit disabled, this method should not be used.

Throws:

TimeoutException

pollMessages

public <K,V> com.google.common.util.concurrent.ListenableFuture<List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>>> pollMessages(String topic,
                                                                                                                                          int numMessagesToPoll,
                                                                                                                                          org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                                                                                                          org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer)

Poll the specified topic for messages

Type Parameters:

K - Type of Key

V - Type of Value

Parameters:

topic - Topic to poll

numMessagesToPoll - Number of messages to read

keyDeserializer - Key deserializer

valueDeserializer - Value deserializer

Returns:

ListenableFuture containing a list of ConsumerRecord objects

pollStringMessages

public com.google.common.util.concurrent.ListenableFuture<List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>>> pollStringMessages(String topic,
                                                                                                                                                    int numMessagesToPoll)

Poll the specified topic for String key-valued messages

Parameters:

topic - Topic to poll

numMessagesToPoll - Number of messages to read

Returns:

ListenableFuture containing a list of ConsumerRecord objects

kafkaLogDir

public Path kafkaLogDir()

Get the Kafka log directory

Returns:

kafka log directory path

kafkaBrokerPort

public int kafkaBrokerPort()

Get the kafka broker port

Returns:

broker port

zookeeperPort

public int zookeeperPort()

Get the zookeeper port

Returns:

zookeeper port

zookeeperConnectionString

public String zookeeperConnectionString()

Get the zookeeper connection string

Returns:

zookeeper connection string