kafka-streams-KafkaStreams.adoc

KafkaStreams — Streams Client

KafkaStreams is the interface for managing and inspecting the execution environment of the processing topology of a Kafka Streams application.

Note	A KafkaStreams instance is also referred by the name streams client (esp. in logs).

A KafkaStreams instance (process) can be started or closed (shut down). The current state is available using state.

KafkaStreams is running when the state is either RUNNING or REBALANCING.

There could be many KafkaStreams instances running simultaneously (e.g. as separate JVM processes each with its own StreamThreads). While a KafkaStreams instance is running, it allows for inspecting streams metadata using allMetadata, allMetadataForStore, and metadataForKey methods.

A KafkaStreams instance exposes monitoring metrics (incl. the Kafka Producer, Consumers, and AdminClient).

Only when in CREATED state, a KafkaStreams instance can be registered with StateRestoreListeners, StateListeners, and UncaughtExceptionHandlers.

Table 1. KafkaStreams API

Method	Description
allMetadata	Collection<StreamsMetadata> allMetadata() StreamsMetadatas for all instances in a multi-instance Kafka Streams application
allMetadataForStore	Collection<StreamsMetadata> allMetadataForStore( String storeName)
cleanUp	void cleanUp() Cleans up the local directory with state stores
close	void close() // (1) boolean close(Duration timeout) Uses Long.MAX_VALUE for the timeout Closes the KafkaStreams instance
localThreadsMetadata	Set<ThreadMetadata> localThreadsMetadata()
metadataForKey	StreamsMetadata metadataForKey( String storeName, K key, Serializer<K> keySerializer) StreamsMetadata metadataForKey( String storeName, K key, StreamPartitioner<? super K, ?> partitioner)
metrics	Map<MetricName, ? extends Metric> metrics() Kafka monitoring metrics of the KafkaStreams instance (incl. the Kafka Producer, Consumers, and AdminClient used by the StreamThreads)
setGlobalStateRestoreListener	void setGlobalStateRestoreListener( StateRestoreListener globalStateRestoreListener) Registers a global StateRestoreListener
setStateListener	void setStateListener( KafkaStreams.StateListener listener) Registers a KafkaStreams.StateListener
setUncaughtExceptionHandler	void setUncaughtExceptionHandler( Thread.UncaughtExceptionHandler eh) Registers a java.lang.UncaughtExceptionHandler
start	void start() Starts the KafkaStreams instance (and the Topology)
state	State state() The current state of the KafkaStreams instance
store	T store( String storeName, QueryableStoreType<T> queryableStoreType)

KafkaStreams is simply a Kafka client that consumes messages from and produces the processing results to Kafka topics (abstracted as SourceNodes and SinkNodes, respectively).

Figure 1. KafkaStreams

Note	A Kafka Streams developer describes the processing logic using a Topology directly (that is a graph of processors) or indirectly through a StreamsBuilder that provides the high-level DSL to define transformations and build a stream processing topology.

val topology: Topology = ...
val config: StreamsConfig = ...

import org.apache.kafka.streams.KafkaStreams
val ks = new KafkaStreams(topology, config)

Once created, KafkaStreams is started up to start consuming, processing, and producing records (as described by a Topology).

ks.start

(only when in CREATED state) KafkaStreams can be given a StateRestoreListener to be informed about the state-related events: onRestoreStart, onBatchRestored and onRestoreEnd (through DelegatingStateRestoreListener).

import org.apache.kafka.streams.processor.StateRestoreListener
val userRestoreListener: StateRestoreListener = ???
ks.setGlobalStateRestoreListener(userRestoreListener)

KafkaStreams uses a InternalTopologyBuilder for the following:

• Creating a StreamsMetadataState, StreamThreads and QueryableStoreProvider

• Requesting a global processor topology (for a GlobalStreamThread)

KafkaStreams uses stream-client [client.id] for the log prefix (with the clientId).

Tip	Enable ALL logging level for org.apache.kafka.streams.KafkaStreams logger to see what happens inside. Add the following line to log4j.properties: log4j.logger.org.apache.kafka.streams.KafkaStreams=ALL Refer to Application Logging Using log4j.

Cleaning Up Local Directory for State Stores — cleanUp Method

void cleanUp()

cleanUp simply requests StateDirectory to clean when KafkaStreams is not running.

Note	cleanUp can only be executed before KafkaStreams will be started or after has been closed.

cleanUp reports a IllegalStateException when KafkaStreams is running.

Cannot clean up while running.

Closing KafkaStreams — close Method

void close()  // (1)
synchronized boolean close(final long timeout, final TimeUnit timeUnit)

1. Calls close(final long timeout, final TimeUnit timeUnit) with 0 timeout

close…​FIXME

Important

Always execute close on a KafkaStreams instance even if you never call start to avoid resource leaks.

Creating KafkaStreams Instance

// public API
KafkaStreams(
  final Topology topology,
  final Properties props) // (1)

// public API (mostly for testing)
KafkaStreams(
  final Topology topology,
  final Properties props,
  final KafkaClientSupplier clientSupplier) // (3)
KafkaStreams(
  final Topology topology,
  final Properties props,
  final Time time)  // (4)

// private/internal API
KafkaStreams(
  final InternalTopologyBuilder internalTopologyBuilder,
  final StreamsConfig config,
  final KafkaClientSupplier clientSupplier) // (5)
KafkaStreams(
  final InternalTopologyBuilder internalTopologyBuilder,
  final StreamsConfig config,
  final KafkaClientSupplier clientSupplier,
  final Time time)  // (6)

1. Calls the internal KafkaStreams (5) with a new DefaultKafkaClientSupplier

2. Calls the internal KafkaStreams (6) with SystemTime

KafkaStreams takes the following to be created:

• InternalTopologyBuilder

• StreamsConfig

• KafkaClientSupplier

• Time

KafkaStreams initializes the internal properties.

While being created, KafkaStreams…​FIXME

KafkaStreams requests the input KafkaClientSupplier for a Kafka AdminClient (for the AdminClient configuration for the clientId).

setRunningFromCreated Internal Method

boolean setRunningFromCreated()

setRunningFromCreated…​FIXME

Note	setRunningFromCreated is used exclusively when KafkaStreams is started.

Starting KafkaStreams — start Method

synchronized void start()
throws IllegalStateException, StreamsException

start starts the Topology (that in turn starts consuming, processing, and producing records).

Internally, start prints out the following DEBUG message to the logs:

Starting Streams client

start marks KafkaStreams as running (i.e. transitions from CREATED to RUNNING state and notifies StateListeners).

start starts global stream thread if defined (which is when…​FIXME)

start starts stream threads.

start schedules a thread that requests StateDirectory to cleanRemovedTasks every state.cleanup.delay.ms milliseconds.

You should see the following DEBUG message in the logs:

Started Streams client

In case the changing state to running fails, start merely prints out the following ERROR message to the logs:

Already stopped, cannot re-start

Registering Global StateRestoreListener — setGlobalStateRestoreListener Method

void setGlobalStateRestoreListener(final StateRestoreListener globalStateRestoreListener)

setGlobalStateRestoreListener registers a StateRestoreListener (in a Kafka Streams application).

Internally, setGlobalStateRestoreListener simply sets the globalStateRestoreListener internal property to be the input StateRestoreListener (only when in CREATED state).

setGlobalStateRestoreListener throws a IllegalStateException when not in CREATED state:

Can only set GlobalStateRestoreListener in CREATED state. Current state is: [state]

allMetadata Method

Collection<StreamsMetadata> allMetadata()

allMetadata makes sure that KafkaStreams is running and requests the StreamsMetadataState for metadata.

Making Sure That KafkaStreams Is Running — validateIsRunning Internal Method

void validateIsRunning()

validateIsRunning throws a IllegalStateException when KafkaStreams is not running. Otherwise, validateIsRunning does nothing.

KafkaStreams is not running. State is [state].

Note	validateIsRunning is used when KafkaStreams is requested to allMetadata, allMetadataForStore, metadataForKey, metadataForKey, store, and localThreadsMetadata.

Internal Properties

Name	Description
clientId	Client ID Used for the following: logging metrics Global thread ID Requesting the KafkaClientSupplier for the global Kafka Consumer (when creating the GlobalStreamThread) Requesting the StreamsConfig for the KafkaAdmin configuration (when creating the AdminClient) Creating StreamThreads The name of the ScheduledExecutorService thread
globalStateRestoreListener	A user-defined global StateRestoreListener to be notified about the state-related events: onRestoreStart, onBatchRestored and onRestoreEnd (through DelegatingStateRestoreListener) Set using setGlobalStateRestoreListener method
globalStreamThread	GlobalStreamThread Initialized exclusively when InternalTopologyBuilder could build a global ProcessorTopology Started when KafkaStreams is being started Set to null while KafkaStreams is being closed
stateDirCleaner	A single-threaded executor (java.util.concurrent.ScheduledExecutorService) that uses a single daemon thread with the name as clientId followed by -CleanupThread Used to schedule a periodic action that requests the StateDirectory to cleanRemovedTasks after and every state.cleanup.delay.ms milliseconds (and only when the state is RUNNING) Initialized when KafkaStreams is created and shut down when requested to close
adminClient	Kafka AdminClient (that allows for managing and inspecting topics, brokers, configurations and ACLs) Initialized when KafkaStreams is created for the only purpose of creating StreamThreads (that simply use it to create a TaskManager) Closed when KafkaStreams is closed
clientId	Client ID that is initialized when KafkaStreams is created as follows: client.id if defined application.id followed by - and the processId
queryableStoreProvider	QueryableStoreProvider
stateDirectory	StateDirectory
stateLock	Object lock for…​FIXME
streamsMetadataState	StreamsMetadataState (with the InternalTopologyBuilder and application.server configuration property) KafkaStreams is simply a public facade to expose the StreamsMetadataState using the following methods: allMetadata allMetadataForStore metadataForKey Initialized when KafkaStreams is created to create StreamThreads
threads	Stream processor threads Note The number of stream processor threads per KafkaStreams instance is controlled by num.stream.threads configuration property (default: 1 processing thread). Created when KafkaStreams is created Started when KafkaStreams is started Shut down when KafkaStreams is closed