<a id="usm-kafka-streams-monitoring"></a>

# Configure Kafka Streams monitoring for Confluent Platform with USM

With Unified Stream Manager (USM), you can monitor Kafka Streams application metrics for
your Confluent Platform clusters in the Confluent Cloud Console. Kafka Streams monitoring uses the same
[KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+Metrics+and+Observability)
client metrics push mechanism as producer and consumer monitoring, where
clients push observability data directly to the Kafka brokers.

The USM Agent collects this observability data and visualizes it in
Confluent Cloud alongside your other cluster metrics.

When you enable client monitoring on your Confluent Platform cluster, the brokers
automatically collect Kafka Streams application metrics in addition to producer
and consumer metrics. You do not need to configure Kafka Streams metrics
separately. Kafka Streams application monitoring requires Confluent Platform 8.2.2 or later
within the 8.2.x series, or Confluent Platform 8.3.0 or later.

To monitor a Kafka Streams application locally through Java Management
Extensions (JMX) instead of USM, see [Monitor Kafka Streams Applications in Confluent Platform](../streams/monitoring.md#streams-monitoring). If
you do not yet have a Kafka Streams application, see the
[Kafka Streams quick start](../streams/quickstart.md#streams-quick-start) or
[Write a Kafka Streams application](../streams/developer-guide/write-streams.md#streams-write-app). For background
on Kafka Streams, see [Kafka Streams for Confluent Platform](../streams/overview.md#kafka-streams).

## Requirements

Before monitoring Kafka Streams applications, ensure your environment meets the
following requirements:

* **Confluent Platform version**: Confluent Platform 8.2.2 or later within the 8.2.x series,
  or Confluent Platform 8.3.0 or later. Earlier Confluent Platform versions collect producer and consumer
  client metrics but not Kafka Streams application metrics in the Confluent Cloud Console.
* **Deployment tool version**: Kafka Streams metrics collection depends on Confluent Platform
  configurations that are available only in recent deployment tool releases.
  If you deploy Confluent Platform with Confluent for Kubernetes (CFK), use CFK 3.3.0 or later. CFK does
  not backport these configurations to earlier releases. If you deploy Confluent Platform
  with Confluent Ansible, use Confluent Ansible 8.2.2 or later.
* **Cluster mode**: Your cluster must run in KRaft mode.
  ZooKeeper-based clusters are not supported.
* **Client versions**: Kafka Streams applications must use a Kafka Streams Java
  library based on Kafka 4.2.0 or later to push metrics to the brokers.
* **Cluster registration**:
  [Register your cluster with Confluent Cloud](get-started.md#usm-start)
  before enabling client monitoring.

## Configure Kafka Streams monitoring

Use one of the following methods to enable client monitoring on your Confluent Platform
cluster. If you already configured client monitoring, the brokers already
collect Kafka Streams application metrics, and you do not need to repeat these
steps. For producer and consumer client monitoring, see
[Configure client monitoring for Confluent Platform with USM](client-monitoring.md#usm-client-monitoring).

### Configure Kafka Streams monitoring with Confluent for Kubernetes (CFK)

Kafka Streams monitoring with CFK requires CFK 3.3.0 or later.

1. Add `clientMetricsEnabled: true` to the `usmAgentClient`
   dependency in both your `Kafka` and `KRaftController`
   resource specs:
   ```yaml
   spec:
     dependencies:
       usmAgentClient:
         url: http://usm-agent.<namespace>.svc.cluster.local:10000
         clientMetricsEnabled: true
   ```
2. Apply the updated configuration. Client metrics cannot be enabled for
   both Control Center and USM simultaneously.
   ```bash
   kubectl apply -f confluent-platform.yaml
   ```

### Configure Kafka Streams monitoring with Confluent Ansible

Kafka Streams monitoring with Confluent Ansible requires Confluent Ansible 8.2.2 or
later.

Set the following in your inventory variables:

```yaml
usm_client_metrics_enabled: true
```

#### NOTE
* `usm_client_metrics_enabled` applies only to Kafka broker and
  KRaft controller hosts.
* You cannot enable client metrics for both Control Center and USM
  simultaneously.

### Configure Kafka Streams monitoring manually

Add the following properties to the `server.properties` file for each Kafka
broker and KRaft controller, then restart the components. The
`confluent.telemetry.external.client.metrics.subscription.metrics.list`
property includes the Kafka Streams application metrics alongside producer and
consumer metrics:

```properties
confluent.telemetry.external.client.metrics.usm.push.enabled=true
confluent.telemetry.external.client.metrics.push.enabled=true
confluent.telemetry.external.client.metrics.delta.temporality=true
confluent.telemetry.external.client.metrics.subscription.interval.ms.list=60000
confluent.telemetry.external.client.metrics.subscription.metrics.list=\
  org.apache.kafka.consumer.fetch.manager.fetch.latency.avg,\
  org.apache.kafka.consumer.connection.creation.total,\
  org.apache.kafka.consumer.fetch.manager.fetch.total,\
  org.apache.kafka.consumer.fetch.manager.bytes.consumed.rate,\
  org.apache.kafka.consumer.request.total,\
  org.apache.kafka.producer.bufferpool.wait.ratio,\
  org.apache.kafka.producer.record.queue.time.avg,\
  org.apache.kafka.producer.request.latency.avg,\
  org.apache.kafka.producer.produce.throttle.time.avg,\
  org.apache.kafka.producer.connection.creation.total,\
  org.apache.kafka.producer.request.total,\
  org.apache.kafka.producer.topic.byte.rate,\
  org.apache.kafka.stream.client.state,\
  org.apache.kafka.stream.thread.thread.state,\
  org.apache.kafka.stream.state.size.all.mem.tables,\
  org.apache.kafka.stream.state.estimate.num.keys,\
  org.apache.kafka.stream.state.block.cache.usage,\
  org.apache.kafka.stream.processor.node.record.e2e.latency.min,\
  org.apache.kafka.stream.processor.node.record.e2e.latency.max,\
  org.apache.kafka.stream.processor.node.record.e2e.latency.avg,\
  org.apache.kafka.stream.thread.poll.ratio,\
  org.apache.kafka.stream.thread.commit.ratio,\
  org.apache.kafka.stream.thread.process.ratio,\
  org.apache.kafka.stream.thread.punctuate.ratio,\
  org.apache.kafka.stream.thread.process.records.avg
```

#### NOTE
Do not enable manual client metrics configuration alongside Control Center.
Doing so causes a delta temporality conflict that results in
inaccurate metrics.

<a id="usm-kafka-streams-metrics-collected"></a>

## Metrics collected for Kafka Streams applications

When Kafka Streams monitoring is enabled, USM collects the following
Kafka Streams application metrics. The
`confluent.telemetry.external.client.metrics.subscription.metrics.list`
property defines which client metrics the brokers collect. To collect a
different set, edit the comma-separated list.

These metrics are a subset of the metrics that a Kafka Streams application
exposes. For the complete list of Kafka Streams metrics and their definitions,
see [Kafka Streams Metrics in Confluent Platform](../streams/kafka-streams-metrics.md#kafka-streams-metrics).

| Metric                                                          | Description                                                                                                                                                                            |
|-----------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `org.apache.kafka.stream.client.state`                          | The state of the Kafka Streams client as a number, which is the<br/>`ordinal()` of the corresponding enum.                                                                             |
| `org.apache.kafka.stream.thread.thread.state`                   | The state of the thread as a number, which is the `ordinal()` of<br/>the corresponding enum.                                                                                           |
| `org.apache.kafka.stream.state.size.all.mem.tables`             | The approximate size of active, unflushed immutable, and pinned<br/>immutable memtables, in bytes.                                                                                     |
| `org.apache.kafka.stream.state.estimate.num.keys`               | The estimated number of keys in the active and unflushed immutable<br/>memtables and storage.                                                                                          |
| `org.apache.kafka.stream.state.block.cache.usage`               | The memory size of the entries residing in block cache, in bytes.                                                                                                                      |
| `org.apache.kafka.stream.processor.node.record.e2e.latency.min` | The minimum end-to-end latency of a record in milliseconds (ms),<br/>measured by comparing the record timestamp with the system time when<br/>it has been fully processed by the node. |
| `org.apache.kafka.stream.processor.node.record.e2e.latency.max` | The maximum end-to-end latency of a record in milliseconds (ms),<br/>measured by comparing the record timestamp with the system time when<br/>it has been fully processed by the node. |
| `org.apache.kafka.stream.processor.node.record.e2e.latency.avg` | The average end-to-end latency of a record in milliseconds (ms),<br/>measured by comparing the record timestamp with the system time when<br/>it has been fully processed by the node. |
| `org.apache.kafka.stream.thread.poll.ratio`                     | The fraction of time the thread has spent on polling records from<br/>the consumer.                                                                                                    |
| `org.apache.kafka.stream.thread.commit.ratio`                   | The fraction of time the thread has spent on committing all tasks.                                                                                                                     |
| `org.apache.kafka.stream.thread.process.ratio`                  | The fraction of time the thread has spent on processing active tasks.                                                                                                                  |
| `org.apache.kafka.stream.thread.punctuate.ratio`                | The fraction of time the thread spends performing punctuating actions<br/>on active tasks.                                                                                             |
| `org.apache.kafka.stream.thread.process.records.avg`            | The average number of records processed within an iteration, computed<br/>as the total count of processed records over the total number of<br/>iterations.                             |

## View Kafka Streams metrics in Confluent Cloud

To view Kafka Streams application metrics, go to the **Kafka Streams** page in
the Confluent Cloud Console for your registered Confluent Platform cluster. For details, see
[Monitor Confluent Platform Kafka Streams Applications with Unified Stream Manager](https://docs.confluent.io/cloud/current/usm/monitor/kafka-streams.html).

## Related content

* [Configure client monitoring for Confluent Platform with USM](client-monitoring.md#usm-client-monitoring)
* [Unified Stream Manager (USM) in Confluent Platform](overview.md#usm-overview)
* [Getting Started with USM](get-started.md#usm-start)
* [Kafka Streams for Confluent Platform](../streams/overview.md#kafka-streams)
* [Monitor Kafka Streams Applications in Confluent Platform](../streams/monitoring.md#streams-monitoring)
* [Kafka Streams Metrics in Confluent Platform](../streams/kafka-streams-metrics.md#kafka-streams-metrics)