Monitor Kafka Streams Applications in Confluent Platform

Monitoring a Kafka Streams application means observing its runtime behavior through metrics that the Kafka Streams library exposes by using Java Management Extensions (JMX), covering client, stream thread, task, processor node, state store, RocksDB, and record cache activity. You can also configure Kafka Streams applications to report these metrics through pluggable reporters such as Confluent Metrics Reporter.

For a complete list of available metrics, see Kafka Streams Metrics.

Access metrics

Access metrics using JMX and reporters

The Kafka Streams library reports a variety of metrics through JMX. Also, you can configure it to report metrics by using additional pluggable stats reporters, which are configured by using the metrics.reporters configuration option, for example:

metrics.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter

The easiest way to view the available metrics is by using tools that enable you to browse JMX MBeans, such as JConsole.

For all Apache Kafka® metrics, see Monitor Kafka with JMX in Confluent Platform.
For Cluster Linking metrics, see Monitor Cluster Metrics and Optimize Links for Cluster Linking on Confluent Platform.

Access metrics programmatically

You can access the entire metrics registry of a Kafka Streams instance read-only through the method KafkaStreams#metrics(). The metrics registry contains all of the available metrics described in this section. For more information, see Kafka Streams Javadocs.

The metrics for Kafka Streams have a four-level hierarchy:

At the top level, there are client-level metrics for each running Kafka Streams client.
Each client has stream threads with their own metrics.
Each stream thread has tasks with their own metrics.
Each task has a number of processor nodes with their own metrics. Also, each task has a number of state stores and record caches, all of which have their own metrics.

Configure metrics granularity

By default, Kafka Streams has metrics with three recording levels: info, debug, and trace. The debug level records most metrics, while the info level records only some of them. The trace level records all possible metrics.

Use the metrics.recording.level configuration option to specify which metrics you want collected, for example:

metrics.recording.level="info"

For more information, see Optional configuration parameters.

Add your own metrics

Application developers using the low-level Processor API can add additional metrics to their application. The ProcessorContext#metrics() method provides a handle to the StreamsMetrics object, which you can use to:

Add latency and throughput metrics by using StreamsMetrics#addLatencyRateTotalSensor() and StreamsMetrics#addRateTotalSensor().
Add any other type of metric by using StreamsMetrics#addSensor().

Runtime status information

Status of Kafka Streams instances

Important

Don’t confuse the runtime state of a KafkaStreams instance, for example, created or rebalancing, with state stores.

A Kafka Streams instance can be in one of several runtime states, as defined in the enum KafkaStreams.State. For example, it might be created but not running; or it might be rebalancing, so its state stores are not available for querying. You can access the current runtime state programmatically using the method KafkaStreams#state(). The documentation of KafkaStreams.State in the Kafka Streams Javadocs lists all the available states.

Also, you can use KafkaStreams#setStateListener() to register a KafkaStreams#StateListener method that is triggered whenever the state changes.

Use the KafkaStreams#metadataForLocalThreads() method to check the runtime state of the current KafkaStreams instance. The metadataForLocalThreads() method returns a ThreadMetadata object for each local stream thread. The ThreadMetadata object describes the runtime state of a thread and the metadata for the thread’s currently assigned tasks.

Get runtime information on Kafka Streams clients

You can get runtime information on these local KafkaStreams clients:

Admin client
Producer clients
Consumer client
Restore consumer client

There is one admin client per KafkaStreams instance, and all other clients are per StreamThread.

Get the names of local KafkaStreams clients by calling the client ID methods on the ThreadMetadata class, like producerClientIds().

Client names are based on a client ID value, which is assigned according to the StreamsConfig.CLIENT_ID_CONFIG and StreamsConfig.APPLICATION_ID_CONFIG configuration settings.

If CLIENT_ID_CONFIG is set, Kafka Streams uses CLIENT_ID_CONFIG for the client ID value.
If CLIENT_ID_CONFIG isn’t set, Kafka Streams uses APPLICATION_ID_CONFIG and appends a random unique identifier (UUID):
```
clientId = StreamsConfig.APPLICATION_ID_CONFIG + "-" + <random-UUID>
```

Kafka Streams creates names for specific clients by appending a thread ID and a descriptive string to the main client ID.

specificClientId = clientId + "-StreamThread-" + <thread-number> + <description>

For example, if CLIENT_ID_CONFIG is set to “MyClientId”, the consumerClientId() method returns a value that resembles MyClientId-StreamThread-2-consumer. If CLIENT_ID_CONFIG isn’t set, and APPLICATION_ID_CONFIG is set to “MyApplicationId”, the consumerClientId() method returns a value that resembles MyApplicationId-8d8ce4a7-85bb-41f7-ac9c-fe6f3cc0959e-StreamThread-2-consumer.

Call the threadName() method to get the thread ID:

threadId = clientId + "-StreamThread-" + <thread-number>

Depending on the configuration settings, an example thread ID resembles MyClientId-StreamThread-2 or MyApplicationId-8d8ce4a7-85bb-41f7-ac9c-fe6f3cc0959e-StreamThread-2.

adminClientId()

Gets the ID of the client application, which is the main client ID value, appended with -admin. Depending on configuration settings, the return value resembles MyClientId-admin or MyApplicationId-8d8ce4a7-85bb-41f7-ac9c-fe6f3cc0959e-admin.

The admin client ID doesn’t contain a thread ID.

producerClientIds()

Gets the names of producer clients. The method returns the thread producer name, which is the main thread ID appended with -producer, for example, MyClientId-StreamThread-2-producer.

Exactly-once semantics version 1 was removed in Kafka Streams 4.0 (Confluent Platform 8.0). In releases that supported it, producerClientIds() returned a Set of task producer names, each with a -<taskId> suffix (for example, MyClientId-StreamThread-2-1_4-producer), where a task ID is a sub-topology ID and a partition number, <subTopologyId>_<partition>, and the subTopologyId is an integer greater than or equal to zero. With exactly-once semantics version 2, or with exactly-once disabled, the method returns the single thread producer name.

For more information, see Stream partitions and tasks.

consumerClientId(): Gets the name of the consumer client. The consumer client name is the main thread ID appended with -consumer, for example, MyClientId-StreamThread-2-consumer.

restoreConsumerClientId(): Gets the name of the restore consumer client. The restore consumer client name is the main thread ID appended with -restore-consumer, for example, MyClientId-StreamThread-2-restore-consumer

Monitor the restoration progress of fault-tolerant state stores

When starting your application, any fault-tolerant state stores don’t need a restoration process, because the persisted state is read from local disk. But there could be situations when a full restore from the backing changelog topic is required, for example, a failure wiped out the local state or your application runs in a stateless environment and persisted data is lost on restarts.

If you have a significant amount of data in the changelog topic, the restoration process could take a non-negligible amount of time. Given that processing of new data won’t start until the restoration process is completed, having a window into the progress of restoration is useful.

To observe the restoration of all state stores, provide your application with an instance of the org.apache.kafka.streams.processor.StateRestoreListener interface. Set the org.apache.kafka.streams.processor.StateRestoreListener by calling the KafkaStreams#setGlobalStateRestoreListener method.

The following code shows a basic implementation example that prints restoration status to the console.

import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.processor.StateRestoreListener;

 public class ConsoleGlobalRestoreListerner implements StateRestoreListener {

    @Override
    public void onRestoreStart(final TopicPartition topicPartition,
                               final String storeName,
                               final long startingOffset,
                               final long endingOffset) {

        System.out.print("Started restoration of " + storeName + " partition " + topicPartition.partition());
        System.out.println(" total records to be restored " + (endingOffset - startingOffset));
    }

    @Override
    public void onBatchRestored(final TopicPartition topicPartition,
                                final String storeName,
                                final long batchEndOffset,
                                final long numRestored) {

        System.out.println("Restored batch " + numRestored + " for " + storeName + " partition " + topicPartition.partition());

    }

    @Override
    public void onRestoreEnd(final TopicPartition topicPartition,
                             final String storeName,
                             final long totalRestored) {

        System.out.println("Restoration complete for " + storeName + " partition " + topicPartition.partition());
    }
}

Important

The StateRestoreListener instance is shared across all org.apache.kafka.streams.processor.internals.StreamThread instances and also used for global stores. Furthermore, all methods are assumed to be stateless. If any stateful operations are desired, then you must provide synchronization internally.

Integration with Confluent Control Center

As of the 3.2 release, Confluent Control Center displays the underlying producer metrics and consumer metrics of a Kafka Streams application, which the Kafka Streams API uses internally whenever data needs to be read from or written to Kafka topics. These metrics can be used, for example, to monitor the “consumer lag” of an application, which indicates whether an application at its current capacity and available computing resources is able to keep up with the incoming data volume.

In Control Center, all of the running instances of a Kafka Streams application appear as a single consumer group.

Restore consumers of an application are displayed separately. Behind the scenes, the Streams API uses a dedicated “restore” consumer for the purposes of fault tolerance and state management. This restore consumer manually assigns and manages the topic partitions it consumes from and is not a member of the application’s consumer group. As a result, the restore consumers are displayed separately from their application.

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.