org.apache.flink.tests.util.kafka

Class LocalStandaloneKafkaResource

java.lang.Object

org.apache.flink.tests.util.kafka.LocalStandaloneKafkaResource

All Implemented Interfaces:

KafkaResource, ExternalResource, org.junit.rules.TestRule

public class LocalStandaloneKafkaResource
extends Object
implements KafkaResource

KafkaResource that downloads kafka and sets up a local kafka cluster with the bundled zookeeper.

Method Summary

Modifier and Type	Method and Description
void	afterTestFailure()
void	afterTestSuccess()
void	before()
void	createTopic(int replicationFactor, int numPartitions, String topic) Creates a topic with the given name, replication factor and number of partitions.
Collection<InetSocketAddress>	getBootstrapServerAddresses() Returns the kafka bootstrap server addresses.
int	getNumPartitions(String topic) Returns the current number of partitions for the given topic.
long	getPartitionOffset(String topic, int partition) Returns the current partition offset for the given partition of the given topic.
InetSocketAddress	getZookeeperAddress() Returns the address of Zookeeper.
List<String>	readMessage(int expectedNumMessages, String groupId, String topic) Reads expectedNumMessages from the given topic.
void	sendKeyedMessages(String topic, String keySeparator, String... messages) Sends the given keyed messages to the given topic.
void	sendMessages(String topic, String... messages) Sends the given messages to the given topic.
void	setNumPartitions(int numPartitions, String topic) Modifies the number of partitions for the given topic.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.tests.util.kafka.KafkaResource

get

Methods inherited from interface org.apache.flink.util.ExternalResource

apply

Method Detail

before

public void before()
            throws Exception

Specified by:

before in interface ExternalResource

Throws:

Exception

afterTestSuccess

public void afterTestSuccess()

Specified by:

afterTestSuccess in interface ExternalResource

afterTestFailure

public void afterTestFailure()

Specified by:

afterTestFailure in interface ExternalResource

createTopic

public void createTopic(int replicationFactor,
                        int numPartitions,
                        String topic)
                 throws IOException

Description copied from interface: KafkaResource

Creates a topic with the given name, replication factor and number of partitions.

Specified by:

createTopic in interface KafkaResource

Parameters:

replicationFactor - replication factor

numPartitions - number of partitions

topic - desired topic name

Throws:

IOException

sendMessages

public void sendMessages(String topic,
                         String... messages)
                  throws IOException

Description copied from interface: KafkaResource

Sends the given messages to the given topic.

Specified by:

sendMessages in interface KafkaResource

Parameters:

topic - topic name

messages - messages to send

Throws:

IOException

sendKeyedMessages

public void sendKeyedMessages(String topic,
                              String keySeparator,
                              String... messages)
                       throws IOException

Description copied from interface: KafkaResource

Sends the given keyed messages to the given topic. The messages themselves should contain the specified keySeparator.

Specified by:

sendKeyedMessages in interface KafkaResource

Parameters:

topic - topic name

keySeparator - the separator used to parse key from value in the messages

messages - messages to send

Throws:

IOException

readMessage

public List<String> readMessage(int expectedNumMessages,
                                String groupId,
                                String topic)
                         throws IOException

Description copied from interface: KafkaResource

Reads expectedNumMessages from the given topic. If we can't read the expected number of messages we throw an exception.

Specified by:

readMessage in interface KafkaResource

Parameters:

expectedNumMessages - expected number of messages that should be read

groupId - group id to identify consumer

topic - topic name

Returns:

read messages

Throws:

IOException

setNumPartitions

public void setNumPartitions(int numPartitions,
                             String topic)
                      throws IOException

Description copied from interface: KafkaResource

Modifies the number of partitions for the given topic.

Specified by:

setNumPartitions in interface KafkaResource

Parameters:

numPartitions - desired number of partitions

topic - topic to modify

Throws:

IOException

getNumPartitions

public int getNumPartitions(String topic)
                     throws IOException

Description copied from interface: KafkaResource

Returns the current number of partitions for the given topic.

Specified by:

getNumPartitions in interface KafkaResource

Parameters:

topic - topic name

Returns:

number of partitions for the given topic

Throws:

IOException

getPartitionOffset

public long getPartitionOffset(String topic,
                               int partition)
                        throws IOException

Description copied from interface: KafkaResource

Returns the current partition offset for the given partition of the given topic.

Specified by:

getPartitionOffset in interface KafkaResource

Parameters:

topic - topic name

partition - topic partition

Returns:

partition offset for the given partition

Throws:

IOException

getBootstrapServerAddresses

public Collection<InetSocketAddress> getBootstrapServerAddresses()

Description copied from interface: KafkaResource

Returns the kafka bootstrap server addresses.

Specified by:

getBootstrapServerAddresses in interface KafkaResource

Returns:

kafka bootstrap server addresses

getZookeeperAddress

public InetSocketAddress getZookeeperAddress()

Description copied from interface: KafkaResource

Returns the address of Zookeeper.

Specified by:

getZookeeperAddress in interface KafkaResource

Returns:

zookeeper address