org.apache.kafka.streams

Class KafkaStreams

java.lang.Object

org.apache.kafka.streams.KafkaStreams

@InterfaceStability.Unstable
public class KafkaStreams
extends Object

Kafka Streams allows for performing continuous computation on input coming from one or more input topics and sends output to zero or more output topics.

The computational logic can be specified either by using the TopologyBuilder class to define the a DAG topology of Processors or by using the KStreamBuilder class which provides the high-level KStream DSL to define the transformation.

The KafkaStreams class manages the lifecycle of a Kafka Streams instance. One stream instance can contain one or more threads specified in the configs for the processing work.

A KafkaStreams instance can co-ordinate with any other instances with the same application ID (whether in this same process, on other processes on this machine, or on remote machines) as a single (possibly distributed) stream processing client. These instances will divide up the work based on the assignment of the input topic partitions so that all partitions are being consumed. If instances are added or failed, all instances will rebalance the partition assignment among themselves to balance processing load.

Internally the KafkaStreams instance contains a normal KafkaProducer and KafkaConsumer instance that is used for reading input and writing output.

A simple example might look like this:

    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-processing-application");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(StreamsConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(StreamsConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(StreamsConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    StreamsConfig config = new StreamsConfig(props);

    KStreamBuilder builder = new KStreamBuilder();
    builder.stream("my-input-topic").mapValues(value -> value.length().toString()).to("my-output-topic");

    KafkaStreams streams = new KafkaStreams(builder, config);
    streams.start();
 

Constructor Summary

Constructors

Constructor and Description
KafkaStreams(TopologyBuilder builder, Properties props) Construct the stream instance.
KafkaStreams(TopologyBuilder builder, StreamsConfig config) Construct the stream instance.
KafkaStreams(TopologyBuilder builder, StreamsConfig config, KafkaClientSupplier clientSupplier) Construct the stream instance.

Method Summary

Methods

Modifier and Type	Method and Description
void	cleanUp() Cleans up local state store directory (state.dir), by deleting all data with regard to the application-id.
void	close() Shutdown this stream instance by signaling all the threads to stop, and then wait for them to join.
void	setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh) Sets the handler invoked when a stream thread abruptly terminates due to an uncaught exception.
void	start() Start the stream instance by starting all its threads.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaStreams

public KafkaStreams(TopologyBuilder builder,
            Properties props)

Construct the stream instance.

Parameters:

builder - the processor topology builder specifying the computational logic

props - properties for the StreamsConfig

KafkaStreams

public KafkaStreams(TopologyBuilder builder,
            StreamsConfig config)

Construct the stream instance.

Parameters:

builder - the processor topology builder specifying the computational logic

config - the stream configs

KafkaStreams

public KafkaStreams(TopologyBuilder builder,
            StreamsConfig config,
            KafkaClientSupplier clientSupplier)

Construct the stream instance.

Parameters:

builder - the processor topology builder specifying the computational logic

config - the stream configs

clientSupplier - the kafka clients supplier which provides underlying producer and consumer clients for this KafkaStreams instance

Method Detail

start

public void start()

Start the stream instance by starting all its threads.

Throws:

IllegalStateException - if process was already started

close

public void close()

Shutdown this stream instance by signaling all the threads to stop, and then wait for them to join.

Throws:

IllegalStateException - if process has not started yet

cleanUp

public void cleanUp()

Cleans up local state store directory (state.dir), by deleting all data with regard to the application-id.

May only be called either before instance is started or after instance is closed.

Throws:

IllegalStateException - if instance is currently running

setUncaughtExceptionHandler

public void setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh)

Sets the handler invoked when a stream thread abruptly terminates due to an uncaught exception.

Parameters:

eh - the object to use as this thread's uncaught exception handler. If null then this thread has no explicit handler.