<a id="usm-client-monitoring"></a>

# Configure client monitoring for Confluent Platform with USM

With Unified Stream Manager (USM), you can enable client monitoring to view producer and
consumer client metrics for your Confluent Platform clusters in the Confluent Cloud Console.
Client monitoring uses the
[KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+Metrics+and+Observability)
client metrics push mechanism, 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.

## Requirements

Before enabling client monitoring, ensure your environment meets
the following requirements:

* **Confluent Platform version**: Confluent Platform 7.9.6 or later within the 7.9.x series,
  or Confluent Platform 8.1.2 or later.
* **Cluster mode**: Your cluster must run in KRaft mode.
  ZooKeeper-based clusters are not supported.
* **Client versions**: Clients must use a supported version to push
  metrics to the brokers:
  * Java client 3.7.2 or later
  * C/C++ client, librdkafka, 2.6 or later
* **Cluster registration**:
  [Register your cluster with Confluent Cloud](get-started.md#usm-start)
  before enabling client monitoring.
* **Python clients**: Python clients do not support all standard
  KIP-714 metrics. Most metric columns appear blank for Python clients
  in the Confluent Cloud Console.

## Configure client monitoring

Enable client monitoring on your Confluent Platform cluster before clients can push metrics
to Confluent Cloud. Use Confluent for Kubernetes (CFK), Confluent Ansible, or manual broker
configuration to set the required broker settings.

### Use Confluent for Kubernetes

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
   ```

### Use Confluent Ansible

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 client monitoring manually

Add the following properties to the `server.properties` file for each Kafka
broker and KRaft controller, then restart the components:

```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-client-metrics-collected"></a>

## Metrics collected by client monitoring

The `confluent.telemetry.external.client.metrics.subscription.metrics.list`
property defines which client metrics the brokers collect. The following
metrics are grouped by client type. To collect a different set, edit the
comma-separated list.

| Client   | Metric                                                          | Description                                                                      |
|----------|-----------------------------------------------------------------|----------------------------------------------------------------------------------|
| Consumer | `org.apache.kafka.consumer.fetch.manager.fetch.latency.avg`     | Average time, in milliseconds, to complete a fetch request.                      |
| Consumer | `org.apache.kafka.consumer.connection.creation.total`           | Total number of broker connections the consumer created.                         |
| Consumer | `org.apache.kafka.consumer.fetch.manager.fetch.total`           | Total number of fetch requests the consumer sent.                                |
| Consumer | `org.apache.kafka.consumer.fetch.manager.bytes.consumed.rate`   | Average number of bytes consumed per second.                                     |
| Consumer | `org.apache.kafka.consumer.request.total`                       | Total number of requests the consumer sent.                                      |
| Producer | `org.apache.kafka.producer.bufferpool.wait.ratio`               | Fraction of time the producer waits for buffer pool space.                       |
| Producer | `org.apache.kafka.producer.record.queue.time.avg`               | Average time, in milliseconds, a record waits in the send buffer.                |
| Producer | `org.apache.kafka.producer.request.latency.avg`                 | Average request latency, in milliseconds.                                        |
| Producer | `org.apache.kafka.producer.produce.throttle.time.avg`           | Average time, in milliseconds, a produce request is throttled by the<br/>broker. |
| Producer | `org.apache.kafka.producer.connection.creation.total`           | Total number of broker connections the producer created.                         |
| Producer | `org.apache.kafka.producer.request.total`                       | Total number of requests the producer sent.                                      |
| Producer | `org.apache.kafka.producer.topic.byte.rate`                     | Average number of bytes sent per second for a topic.                             |
| Streams  | `org.apache.kafka.stream.client.state`                          | Current state of the Kafka Streams client.                                       |
| Streams  | `org.apache.kafka.stream.thread.thread.state`                   | Current state of a stream thread.                                                |
| Streams  | `org.apache.kafka.stream.state.size.all.mem.tables`             | Total size, in bytes, of all in-memory state store tables.                       |
| Streams  | `org.apache.kafka.stream.state.estimate.num.keys`               | Estimated number of keys in the state store.                                     |
| Streams  | `org.apache.kafka.stream.state.block.cache.usage`               | Block cache memory, in bytes, the state store uses.                              |
| Streams  | `org.apache.kafka.stream.processor.node.record.e2e.latency.min` | Minimum end-to-end latency of a record at a processor node.                      |
| Streams  | `org.apache.kafka.stream.processor.node.record.e2e.latency.max` | Maximum end-to-end latency of a record at a processor node.                      |
| Streams  | `org.apache.kafka.stream.processor.node.record.e2e.latency.avg` | Average end-to-end latency of a record at a processor node.                      |
| Streams  | `org.apache.kafka.stream.thread.poll.ratio`                     | Fraction of time the stream thread spends polling records.                       |
| Streams  | `org.apache.kafka.stream.thread.commit.ratio`                   | Fraction of time the stream thread spends committing.                            |
| Streams  | `org.apache.kafka.stream.thread.process.ratio`                  | Fraction of time the stream thread spends processing.                            |
| Streams  | `org.apache.kafka.stream.thread.punctuate.ratio`                | Fraction of time the stream thread spends on punctuation.                        |
| Streams  | `org.apache.kafka.stream.thread.process.records.avg`            | Average number of records processed per call.                                    |

## View client metrics in Confluent Cloud

To view producer and consumer metrics, go to the **Clients** page in
the Confluent Cloud Console for your registered Confluent Platform cluster. For details, see
[Monitor Confluent Platform clients](https://docs.confluent.io/cloud/current/usm/monitor/clients.html).

## Related content

* [Unified Stream Manager (USM) in Confluent Platform](overview.md#usm-overview)
* [Getting Started with USM](get-started.md#usm-start)