com.github.charithe.kafka

Class EphemeralKafkaBroker

java.lang.Object

com.github.charithe.kafka.EphemeralKafkaBroker

public class EphemeralKafkaBroker
extends Object

Method Summary

Modifier and Type	Method and Description
Properties	consumerConfig() Create a minimal consumer configuration with auto commit enabled.
Properties	consumerConfig(boolean enableAutoCommit) Create a minimal consumer configuration.
static EphemeralKafkaBroker	create() Create a new ephemeral Kafka broker with random broker port and Zookeeper port
static EphemeralKafkaBroker	create(int kafkaPort) Create a new ephemeral Kafka broker with the specified broker port and random Zookeeper port
static EphemeralKafkaBroker	create(int kafkaPort, int zookeeperPort) Create a new ephemeral Kafka broker with the specified broker port and Zookeeper port
static EphemeralKafkaBroker	create(int kafkaPort, int zookeeperPort, Properties overrideBrokerProperties) Create a new ephemeral Kafka broker with the specified broker port, Zookeeper port and config overrides.
<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, Properties overrideConfig) Create a consumer that can read from this broker
<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, Properties overrideConfig) Create a producer that can write to this broker
Optional<String>	getBrokerList() Get the current broker list string
Optional<Integer>	getKafkaPort() Get the broker port
Optional<String>	getLogDir() Get the path to the Kafka log directory
Optional<String>	getZookeeperConnectString() Get the current Zookeeper connection string
Optional<Integer>	getZookeeperPort() Get the Zookeeper port
boolean	isRunning() Is the broker running?
Properties	producerConfig() Create a minimal producer configuration that can be used to produce to this broker
CompletableFuture<Void>	start() Start the Kafka broker
void	stop() Stop the Kafka broker

Methods inherited from class java.lang.Object

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

Method Detail

create

public static EphemeralKafkaBroker create()

Create a new ephemeral Kafka broker with random broker port and Zookeeper port

Returns:

EphemeralKafkaBroker

create

public static EphemeralKafkaBroker create(int kafkaPort)

Create a new ephemeral Kafka broker with the specified broker port and random Zookeeper port

Parameters:

kafkaPort - Port the broker should listen on

Returns:

EphemeralKafkaBroker

create

public static EphemeralKafkaBroker create(int kafkaPort,
                                          int zookeeperPort)

Create a new ephemeral Kafka broker with the specified broker port and Zookeeper port

Parameters:

kafkaPort - Port the broker should listen on

zookeeperPort - Port the Zookeeper should listen on

Returns:

EphemeralKafkaBroker

create

public static EphemeralKafkaBroker create(int kafkaPort,
                                          int zookeeperPort,
                                          Properties overrideBrokerProperties)

Create a new ephemeral Kafka broker with the specified broker port, Zookeeper port and config overrides.

Parameters:

kafkaPort - Port the broker should listen on

zookeeperPort - Port the Zookeeper should listen on

overrideBrokerProperties - Broker properties to override. Pass null if there aren't any.

Returns:

EphemeralKafkaBroker

start

public CompletableFuture<Void> start()
                              throws Exception

Start the Kafka broker

Throws:

Exception

stop

public void stop()

Stop the Kafka broker

producerConfig

public Properties producerConfig()

Create a minimal producer configuration that can be used to produce to this broker

Returns:

Properties

consumerConfig

public Properties consumerConfig()

Create a minimal consumer configuration with auto commit enabled. Offset is set to "earliest".

Returns:

Properies

consumerConfig

public Properties consumerConfig(boolean enableAutoCommit)

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

Returns:

Properties

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,
                                                                                 Properties overrideConfig)

Create a producer that can write to this broker

Type Parameters:

K - Type of Key

V - Type of Value

Parameters:

keySerializer - Key serializer class

valueSerializer - Valuer serializer class

overrideConfig - Producer config to override. Pass null if there aren't any.

Returns:

KafkaProducer

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,
                                                                                 Properties overrideConfig)

Create a consumer that can read from this broker

Type Parameters:

K - Type of Key

V - Type of Value

Parameters:

keyDeserializer - Key deserializer

valueDeserializer - Value deserializer

overrideConfig - Consumer config to override. Pass null if there aren't any

Returns:

KafkaConsumer

getKafkaPort

public Optional<Integer> getKafkaPort()

Get the broker port

Returns:

An optional that will only contain a value if the broker is running

getZookeeperPort

public Optional<Integer> getZookeeperPort()

Get the Zookeeper port

Returns:

An optional that will only contain a value if the broker is running

getLogDir

public Optional<String> getLogDir()

Get the path to the Kafka log directory

Returns:

An Optional that will only contain a value if the broker is running

getZookeeperConnectString

public Optional<String> getZookeeperConnectString()

Get the current Zookeeper connection string

Returns:

An Optional that will only contain a value if the broker is running

getBrokerList

public Optional<String> getBrokerList()

Get the current broker list string

Returns:

An Optional that will only contain a value if the broker is running

isRunning

public boolean isRunning()

Is the broker running?

Returns:

True if the broker is running