org.apache.kafka.streams

Class KafkaStreams

java.lang.Object

org.apache.kafka.streams.KafkaStreams

@InterfaceStability.Unstable
public class KafkaStreams
extends Object

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

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

One KafkaStreams 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 the same process, on other processes on this machine, or on remote machines) as a single (possibly distributed) stream processing application. 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 fail, all (remaining) instances will rebalance the partition assignment among themselves to balance processing load and ensure that all input topic partitions are processed.

Internally a 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_SERDE_CLASS_CONFIG, Serdes.String().getClass());
 props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
 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();
 

See Also:

KStreamBuilder, TopologyBuilder

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaStreams.State Kafka Streams states are the possible state that a Kafka Streams instance can be in.
static interface	KafkaStreams.StateListener Listen to KafkaStreams.State change events.

Constructor Summary

Constructors

Constructor and Description
KafkaStreams(TopologyBuilder builder, Properties props) Create a KafkaStreams instance.
KafkaStreams(TopologyBuilder builder, StreamsConfig config) Create a KafkaStreams instance.
KafkaStreams(TopologyBuilder builder, StreamsConfig config, KafkaClientSupplier clientSupplier) Create a KafkaStreams instance.

Method Summary

Methods

Modifier and Type	Method and Description
Collection<StreamsMetadata>	allMetadata() Find all currently running KafkaStreams instances (potentially remotely) that use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application) and return StreamsMetadata for each discovered instance.
Collection<StreamsMetadata>	allMetadataForStore(String storeName) Find all currently running KafkaStreams instances (potentially remotely) that use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application) and that contain a StateStore with the given storeName and return StreamsMetadata for each discovered instance.
void	cleanUp() Do a clean up of the local StateStore directory (StreamsConfig.STATE_DIR_CONFIG) by deleting all data with regard to the application ID.
void	close() Shutdown this KafkaStreams instance by signaling all the threads to stop, and then wait for them to join.
boolean	close(long timeout, TimeUnit timeUnit) Shutdown this KafkaStreams by signaling all the threads to stop, and then wait up to the timeout for the threads to join.
<K> StreamsMetadata	metadataForKey(String storeName, K key, Serializer<K> keySerializer) Find the currently running KafkaStreams instance (potentially remotely) that use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application) and that contain a StateStore with the given storeName and the StateStore contains the given key and return StreamsMetadata for it.
<K> StreamsMetadata	metadataForKey(String storeName, K key, StreamPartitioner<? super K,?> partitioner) Find the currently running KafkaStreams instance (potentially remotely) that use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application) and that contain a StateStore with the given storeName and the StateStore contains the given key and return StreamsMetadata for it.
Map<MetricName,? extends Metric>	metrics() Get read-only handle on global metrics registry.
void	setStateListener(KafkaStreams.StateListener listener) An app can set a single KafkaStreams.StateListener so that the app is notified when state changes.
void	setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh) Set the handler invoked when a internal thread abruptly terminates due to an uncaught exception.
void	start() Start the KafkaStreams instance by starting all its threads.
KafkaStreams.State	state() Return the current KafkaStreams.State of this KafkaStreams instance.
<T> T	store(String storeName, QueryableStoreType<T> queryableStoreType) Get a facade wrapping the local StateStore instances with the provided storeName if the Store's type is accepted by the provided queryableStoreType.
String	toString() Produce a string representation containing useful information about this KafkaStream instance such as thread IDs, task IDs, and a representation of the topology DAG including StateStores (cf.
String	toString(String indent) Produce a string representation containing useful information about this KafkaStream instance such as thread IDs, task IDs, and a representation of the topology DAG including StateStores (cf.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaStreams

public KafkaStreams(TopologyBuilder builder,
            Properties props)

Create a KafkaStreams instance.

Parameters:

builder - the processor topology builder specifying the computational logic

props - properties for StreamsConfig

KafkaStreams

public KafkaStreams(TopologyBuilder builder,
            StreamsConfig config)

Create a KafkaStreams instance.

Parameters:

builder - the processor topology builder specifying the computational logic

config - the Kafka Streams configuration

KafkaStreams

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

Create a KafkaStreams instance.

Parameters:

builder - the processor topology builder specifying the computational logic

config - the Kafka Streams configuration

clientSupplier - the Kafka clients supplier which provides underlying producer and consumer clients for the new KafkaStreams instance

Method Detail

setStateListener

public void setStateListener(KafkaStreams.StateListener listener)

An app can set a single KafkaStreams.StateListener so that the app is notified when state changes.

Parameters:

listener - a new state listener

state

public KafkaStreams.State state()

Return the current KafkaStreams.State of this KafkaStreams instance.

Returns:

the currnt state of this Kafka Streams instance

metrics

public Map<MetricName,? extends Metric> metrics()

Get read-only handle on global metrics registry.

Returns:

Map of all metrics.

start

public void start()
           throws IllegalStateException,
                  StreamsException

Start the KafkaStreams instance by starting all its threads.

Note, for brokers with version 0.9.x or lower, the broker version cannot be checked. There will be no error and the client will hang and retry to verify the broker version until it times out.

Throws:

IllegalStateException - if process was already started

StreamsException - if the Kafka brokers have version 0.10.0.x

close

public void close()

Shutdown this KafkaStreams instance by signaling all the threads to stop, and then wait for them to join. This will block until all threads have stopped.

close

public boolean close(long timeout,
            TimeUnit timeUnit)

Shutdown this KafkaStreams by signaling all the threads to stop, and then wait up to the timeout for the threads to join. A timeout of 0 means to wait forever.

Parameters:

timeout - how long to wait for the threads to shutdown

timeUnit - unit of time used for timeout

Returns:

true if all threads were successfully stopped—false if the timeout was reached before all threads stopped

toString

public String toString()

Produce a string representation containing useful information about this KafkaStream instance such as thread IDs, task IDs, and a representation of the topology DAG including StateStores (cf. TopologyBuilder and KStreamBuilder).

Overrides:

toString in class Object

Returns:

A string representation of the Kafka Streams instance.

toString

public String toString(String indent)

Produce a string representation containing useful information about this KafkaStream instance such as thread IDs, task IDs, and a representation of the topology DAG including StateStores (cf. TopologyBuilder and KStreamBuilder).

Parameters:

indent - the top-level indent for each line

Returns:

A string representation of the Kafka Streams instance.

cleanUp

public void cleanUp()

Do a clean up of the local StateStore directory (StreamsConfig.STATE_DIR_CONFIG) by deleting all data with regard to the application ID.

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

Calling this method triggers a restore of local StateStores on the next application start.

Throws:

IllegalStateException - if the instance is currently running

setUncaughtExceptionHandler

public void setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh)

Set the handler invoked when a internal thread abruptly terminates due to an uncaught exception.

Parameters:

eh - the uncaught exception handler for all internal threads; null deletes the current handler

allMetadata

public Collection<StreamsMetadata> allMetadata()

Find all currently running KafkaStreams instances (potentially remotely) that use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application) and return StreamsMetadata for each discovered instance.

Note: this is a point in time view and it may change due to partition reassignment.

Returns:

StreamsMetadata for each KafkaStreams instances of this application

allMetadataForStore

public Collection<StreamsMetadata> allMetadataForStore(String storeName)

Find all currently running KafkaStreams instances (potentially remotely) that

• use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application)
• and that contain a StateStore with the given storeName

and return StreamsMetadata for each discovered instance.

Note: this is a point in time view and it may change due to partition reassignment.

Parameters:

storeName - the storeName to find metadata for

Returns:

StreamsMetadata for each KafkaStreams instances with the provide storeName of this application

metadataForKey

public <K> StreamsMetadata metadataForKey(String storeName,
                                 K key,
                                 Serializer<K> keySerializer)

Find the currently running KafkaStreams instance (potentially remotely) that

• use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application)
• and that contain a StateStore with the given storeName
• and the StateStore contains the given key

and return StreamsMetadata for it.

This will use the default Kafka Streams partitioner to locate the partition. If a custom partitioner has been configured via StreamsConfig, KStream.through(StreamPartitioner, String), or KTable.through(StreamPartitioner, String, String), or if the original KTable's input topic is partitioned differently, please use metadataForKey(String, Object, StreamPartitioner).

Note:

• this is a point in time view and it may change due to partition reassignment
• the key may not exist in the StateStore; this method provides a way of finding which host it would exist on

Type Parameters:

K - key type

Parameters:

storeName - the storeName to find metadata for

key - the key to find metadata for

keySerializer - serializer for the key

Returns:

StreamsMetadata for the KafkaStreams instance with the provide storeName and key of this application or StreamsMetadata.NOT_AVAILABLE if Kafka Streams is (re-)initializing

metadataForKey

public <K> StreamsMetadata metadataForKey(String storeName,
                                 K key,
                                 StreamPartitioner<? super K,?> partitioner)

Find the currently running KafkaStreams instance (potentially remotely) that

• use the same application ID as this instance (i.e., all instances that belong to the same Kafka Streams application)
• and that contain a StateStore with the given storeName
• and the StateStore contains the given key

and return StreamsMetadata for it.

Note:

• this is a point in time view and it may change due to partition reassignment
• the key may not exist in the StateStore; this method provides a way of finding which host it would exist on

Type Parameters:

K - key type

Parameters:

storeName - the storeName to find metadata for

key - the key to find metadata for

partitioner - the partitioner to be use to locate the host for the key

Returns:

StreamsMetadata for the KafkaStreams instance with the provide storeName and key of this application or StreamsMetadata.NOT_AVAILABLE if Kafka Streams is (re-)initializing

store

public <T> T store(String storeName,
          QueryableStoreType<T> queryableStoreType)

Get a facade wrapping the local StateStore instances with the provided storeName if the Store's type is accepted by the provided queryableStoreType. The returned object can be used to query the StateStore instances.

Type Parameters:

T - return type

Parameters:

storeName - name of the store to find

queryableStoreType - accept only stores that are accepted by QueryableStoreType.accepts(StateStore)

Returns:

A facade wrapping the local StateStore instances

Throws:

InvalidStateStoreException - if Kafka Streams is (re-)initializing or a store with storeName and queryableStoreType doesnt' exist