org.apache.beam.it.kafka

Class KafkaResourceManager

java.lang.Object

org.apache.beam.it.testcontainers.TestContainerResourceManager<GenericContainer<?>>

org.apache.beam.it.kafka.KafkaResourceManager

All Implemented Interfaces:

org.apache.beam.it.common.ResourceManager

public class KafkaResourceManager
extends TestContainerResourceManager<GenericContainer<?>>
implements org.apache.beam.it.common.ResourceManager

Client for managing Kafka resources.

The class supports multiple topic names per server object. A topic is created if one has not been created already.

The topic name is formed using testId. The topic name will be "{testId}-{ISO8601 time, microsecond precision}", with additional formatting.

The class is thread-safe.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaResourceManager.Builder Builder for KafkaResourceManager.

Field Summary

Fields inherited from class org.apache.beam.it.testcontainers.TestContainerResourceManager

HOST_IP, port

Method Summary

Modifier and Type	Method and Description
<K,V> KafkaConsumer<K,V>	buildConsumer(Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) Build a KafkaConsumer for the given serializer and deserializers.
static KafkaResourceManager.Builder	builder(java.lang.String testId)
<K,V> KafkaProducer<K,V>	buildProducer(Serializer<K> keySerializer, Serializer<V> valueSerializer) Build a KafkaProducer for the given serializer and deserializers.
void	cleanupAll() Deletes all created resources and cleans up the Kafka client, making the manager object unusable.
java.lang.String	createTopic(java.lang.String topicName, int partitions) Creates a kafka topic.
java.lang.String	getBootstrapServers() Returns the kafka bootstrap server connection string.
java.util.Set<java.lang.String>	getTopicNames() Returns a list of names of the topics that this kafka manager will operate in.

Methods inherited from class org.apache.beam.it.testcontainers.TestContainerResourceManager

getDockerImageName, getHost, getPort

Methods inherited from class java.lang.Object

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

Method Detail

builder

public static KafkaResourceManager.Builder builder(java.lang.String testId)

getBootstrapServers

public java.lang.String getBootstrapServers()

Returns the kafka bootstrap server connection string.

getTopicNames

public java.util.Set<java.lang.String> getTopicNames()

Returns a list of names of the topics that this kafka manager will operate in.

Returns:

names of the kafka topics.

createTopic

public java.lang.String createTopic(java.lang.String topicName,
                                    int partitions)
                             throws KafkaResourceManagerException

Creates a kafka topic.

Note: Implementations may do topic creation here, if one does not already exist.

Parameters:

topicName - Topic name to associate with the given kafka instance.

partitions - Number of partitions on the topic.

Returns:

The name of the topic that was created.

Throws:

KafkaResourceManagerException - if there is an error creating the kafka topic.

buildProducer

public <K,V> KafkaProducer<K,V> buildProducer(Serializer<K> keySerializer,
                                              Serializer<V> valueSerializer)

Build a KafkaProducer for the given serializer and deserializers.

buildConsumer

public <K,V> KafkaConsumer<K,V> buildConsumer(Deserializer<K> keyDeserializer,
                                              Deserializer<V> valueDeserializer)

Build a KafkaConsumer for the given serializer and deserializers.

cleanupAll

public void cleanupAll()
                throws KafkaResourceManagerException

Deletes all created resources and cleans up the Kafka client, making the manager object unusable.

Specified by:

cleanupAll in interface org.apache.beam.it.common.ResourceManager

Overrides:

cleanupAll in class TestContainerResourceManager<GenericContainer<?>>

Throws:

KafkaResourceManagerException - if there is an error deleting the Kafka resources.