<a id="rust-client-observability"></a>

# Observability for the Kafka Rust Client

The Rust client for Apache Kafka® includes built-in logging and a metrics framework
for monitoring client behavior and performance. Use these logs and metrics
for troubleshooting.

<a id="rust-client-log-levels"></a>

## Log levels

The `log` crate defines the following levels, from least to most verbose:

| Level   | When the client uses it                                                                                          |
|---------|------------------------------------------------------------------------------------------------------------------|
| `error` | Unrecoverable conditions that prevent normal operation.                                                          |
| `warn`  | Recoverable problems, unexpected configuration values, and deprecated<br/>behavior.                              |
| `info`  | High-level lifecycle events such as client creation and shutdown.                                                |
| `debug` | Per-request activity, broker connections, and internal state changes<br/>useful for diagnosing problems.         |
| `trace` | Fine-grained detail including network I/O and individual record<br/>handling. Produces a large volume of output. |

<a id="rust-client-logging"></a>

## Set up logging

To set up logging, add the [log crate](https://crates.io/crates/log),
`log = "0.4"`, to your project. Set up a concrete logging backend to
control how and where the output appears.

[env_logger](https://crates.io/crates/env_logger) is a common
choice for command-line applications and examples.

1. Add it to your project:
   ```bash
   cargo add env_logger
   ```
2. Initialize `env_logger` at the start of your application, before creating
   your client:
   ```rust
   fn main() {
       env_logger::init();
       // create your producer, consumer, or admin client here
   }
   ```
3. Before running your application, set the `RUST_LOG` environment variable
   to control which log records appear:
   ```bash
   RUST_LOG=confluent_kafka=debug cargo run
   ```

   Alternatively, set the level for only the client crate:
   ```bash
   RUST_LOG=confluent_kafka=info cargo run
   ```

   You can also combine the log level with your application’s own logging:
   ```bash
   RUST_LOG=my_app=debug,confluent_kafka=warn cargo run
   ```

For the full syntax, see the
[env_logger documentation](https://docs.rs/env_logger).

<a id="rust-client-metrics"></a>

## Metrics

The client tracks performance and operational data in an internal metrics
framework that uses the same design as the Kafka Java Client. The framework uses
these terms:

Metric
: A single named measurement, such as an average fetch size. Each metric is
  identified by a [MetricName](#rust-client-metric-name) that carries a
  name, a logical group, a description, and a map of key-value tags.

Sensor
: A component that records a continuous sequence of numerical values and
  updates a set of associated metrics with statistics such as averages,
  maximums, and minimums.

Recording level
: A per-sensor setting that controls whether the sensor’s recordings are
  collected, based on the configured `metrics.recording.level`. All built-in
  sensors use `INFO`-level recording.

Registry
: The client’s internal collection of all currently registered metrics. You
  read the registry through `metrics()`, as described in
  [Accessing metrics](#rust-client-accessing-metrics).

<a id="rust-client-accessing-metrics"></a>

### Accessing metrics

Call `metrics()` on a `KafkaConsumer` or `KafkaProducer` to get a
live view of all currently registered metrics:

```rust
use confluent_kafka::common::Metric;
use confluent_kafka::consumer::Consumer;

let view = consumer.metrics();
for (name, metric) in view.snapshot().iter() {
    println!("{}: {:?}", name, metric.metric_value());
}
```

`metrics()` returns a `MetricsView`, which is a live, read-only view of
the client’s metrics registry. Every lookup reads the registry as it is at the
time of the call. To iterate over all metrics or take a point-in-time copy, call
`MetricsView::snapshot()`. This returns an
`Arc<HashMap<MetricName, Arc<KafkaMetric>>>` reflecting the registry at that
moment. Metrics added or removed afterward appear in `MetricsView`, but
not in `MetricsView::snapshot()`.

<a id="rust-client-metric-name"></a>

### MetricName

Each metric is identified by a `MetricName` with these fields:

| Field         | Description                                                                                 |
|---------------|---------------------------------------------------------------------------------------------|
| `name`        | The metric’s short name. For example `fetch-size-avg`.                                      |
| `group`       | The logical group the metric belongs to. For example,<br/>`consumer-fetch-manager-metrics`. |
| `description` | A human-readable description of what the metric measures.                                   |
| `tags`        | Additional key-value attributes, such as `client-id` or `topic`.                            |

Two `MetricName` values are equal when their `name`, `group`, and
`tags` match. Equality ignores `description`, matching the behavior of the
Java Client.

<a id="rust-client-metric-value"></a>

### MetricValue

`metric_value()` on a `KafkaMetric` returns a `MetricValue` enum:

| Variant          | Description                                             |
|------------------|---------------------------------------------------------|
| `Double(f64)`    | A measurable value such as a rate, average, or maximum. |
| `String(String)` | A string-valued gauge.                                  |
| `Long(i64)`      | A long-valued gauge.                                    |
| `Int(i32)`       | An integer-valued gauge.                                |

<a id="rust-client-consumer-metrics"></a>

### Consumer metric groups

The consumer registers metrics in several groups. `KafkaAdminClient`
doesn’t expose a `metrics()` method. All metrics listed here use
`INFO`-level recording and are always active regardless of the
`metrics.recording.level` setting.

#### Consumer metrics

The `consumer-metrics` cover poll timing, commit timing, and idle ratios.

| Metric name                 | Description                                                                                                         |
|-----------------------------|---------------------------------------------------------------------------------------------------------------------|
| `last-poll-seconds-ago`     | The number of seconds since the last `poll()` invocation. If<br/>`poll()` has never been called, returns `-1`.      |
| `time-between-poll-avg`     | The average delay between `poll()` invocations, in milliseconds.                                                    |
| `time-between-poll-max`     | The maximum delay between `poll()` invocations, in milliseconds.                                                    |
| `poll-idle-ratio-avg`       | The average fraction of time that `poll()` is idle, as opposed to<br/>waiting for the user code to process records. |
| `commit-sync-time-ns-total` | The total time spent in `commit_sync`, in nanoseconds.                                                              |
| `committed-time-ns-total`   | The total time spent in `committed`, in nanoseconds.                                                                |

#### Fetch manager metrics

The `consumer-fetch-manager-metrics` track fetch request activity
and record consumption at the client, topic, and partition level.

#### Client-level fetch metrics

| Metric name               | Description                                                                                                                                             |
|---------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------|
| `fetch-size-avg`          | The average number of bytes fetched per request.                                                                                                        |
| `fetch-size-max`          | The maximum number of bytes fetched per request.                                                                                                        |
| `bytes-consumed-rate`     | The average number of bytes consumed per second.                                                                                                        |
| `bytes-consumed-total`    | The total number of bytes consumed.                                                                                                                     |
| `records-per-request-avg` | The average number of records per fetch request.                                                                                                        |
| `records-consumed-rate`   | The average number of records consumed per second.                                                                                                      |
| `records-consumed-total`  | The total number of records consumed.                                                                                                                   |
| `fetch-latency-avg`       | The average time taken for a fetch request, in milliseconds.                                                                                            |
| `fetch-latency-max`       | The maximum time taken for a fetch request, in milliseconds.                                                                                            |
| `fetch-rate`              | The number of fetch requests per second.                                                                                                                |
| `fetch-total`             | The total number of fetch requests.                                                                                                                     |
| `records-lag-max`         | The maximum lag, in number of records, for any partition in<br/>the current window. Lag is based on the current position, not the<br/>committed offset. |
| `records-lead-min`        | The minimum lead, in number of records, for any partition in<br/>the current window.                                                                    |
| `fetch-throttle-time-avg` | The average throttle time imposed by the broker, in milliseconds.                                                                                       |
| `fetch-throttle-time-max` | The maximum throttle time imposed by the broker, in milliseconds.                                                                                       |

Topic-level fetch metrics use the same names as the client-level metrics
and are distinguished by the `topic` tag.

#### Topic-level fetch metrics

| Metric name               | Description                                                       |
|---------------------------|-------------------------------------------------------------------|
| `fetch-size-avg`          | The average number of bytes fetched per request for this topic.   |
| `fetch-size-max`          | The maximum number of bytes fetched per request for this topic.   |
| `bytes-consumed-rate`     | The average number of bytes consumed per second for this topic.   |
| `bytes-consumed-total`    | The total number of bytes consumed for this topic.                |
| `records-per-request-avg` | The average number of records per fetch request for this topic.   |
| `records-consumed-rate`   | The average number of records consumed per second for this topic. |
| `records-consumed-total`  | The total number of records consumed for this topic.              |

Partition-level fetch metrics are tagged with `topic` and `partition`.

#### Partition-level fetch metrics

| Metric name              | Description                                                                         |
|--------------------------|-------------------------------------------------------------------------------------|
| `records-lag`            | The latest lag of the partition.                                                    |
| `records-lag-max`        | The maximum lag of the partition.                                                   |
| `records-lag-avg`        | The average lag of the partition.                                                   |
| `records-lead`           | The latest lead of the partition.                                                   |
| `records-lead-min`       | The minimum lead of the partition.                                                  |
| `records-lead-avg`       | The average lead of the partition.                                                  |
| `preferred-read-replica` | The current read replica for the partition, or `-1` if reading from<br/>the leader. |

#### Rebalance metrics

The `consumer-coordinator-metrics` track group rebalance frequency,
latency, and failure rate.

| Metric name                      | Description                                                                      |
|----------------------------------|----------------------------------------------------------------------------------|
| `rebalance-latency-avg`          | The average time taken for a group rebalance, in milliseconds.                   |
| `rebalance-latency-max`          | The maximum time taken for a group rebalance, in milliseconds.                   |
| `rebalance-latency-total`        | The total time spent in group rebalances, in milliseconds.                       |
| `rebalance-total`                | The total number of rebalance events.                                            |
| `rebalance-rate-per-hour`        | The number of rebalance events per hour.                                         |
| `failed-rebalance-total`         | The total number of failed rebalance events.                                     |
| `failed-rebalance-rate-per-hour` | The number of failed rebalance events per hour.                                  |
| `last-rebalance-seconds-ago`     | The number of seconds since the last rebalance event.                            |
| `assigned-partitions`            | The number of partitions currently assigned to this consumer.                    |
| `partition-assigned-latency-avg` | The average time taken for a partition-assigned rebalance listener<br/>callback. |
| `partition-assigned-latency-max` | The maximum time taken for a partition-assigned rebalance listener<br/>callback. |
| `partition-revoked-latency-avg`  | The average time taken for a partition-revoked rebalance listener<br/>callback.  |
| `partition-revoked-latency-max`  | The maximum time taken for a partition-revoked rebalance listener<br/>callback.  |
| `partition-lost-latency-avg`     | The average time taken for a partition-lost rebalance listener<br/>callback.     |
| `partition-lost-latency-max`     | The maximum time taken for a partition-lost rebalance listener<br/>callback.     |

#### Heartbeat metrics

The `consumer-coordinator-metrics` track the frequency, latency,
and recency of heartbeats sent to the group coordinator.

| Metric name                   | Description                                                                  |
|-------------------------------|------------------------------------------------------------------------------|
| `heartbeat-response-time-max` | The maximum time taken to receive a heartbeat response, in<br/>milliseconds. |
| `heartbeat-rate`              | The number of heartbeats sent per second.                                    |
| `heartbeat-total`             | The total number of heartbeats sent.                                         |
| `last-heartbeat-seconds-ago`  | The number of seconds since the last heartbeat.                              |

#### Commit metrics

The `consumer-coordinator-metrics` track commit requests sent
to the group coordinator.

| Metric name          | Description                                                   |
|----------------------|---------------------------------------------------------------|
| `commit-latency-avg` | The average time taken for a commit request, in milliseconds. |
| `commit-latency-max` | The maximum time taken for a commit request, in milliseconds. |
| `commit-rate`        | The number of commit calls per second.                        |
| `commit-total`       | The total number of commit calls.                             |

#### Internal consumer thread metrics

The `consumer-metrics` track the internal async event queues
between the application thread and the consumer’s background network thread.

| Metric name                                   | Description                                                                                 |
|-----------------------------------------------|---------------------------------------------------------------------------------------------|
| `time-between-network-thread-poll-avg`        | The average time between network thread poll iterations, in<br/>milliseconds.               |
| `time-between-network-thread-poll-max`        | The maximum time between network thread poll iterations, in<br/>milliseconds.               |
| `application-event-queue-size`                | The current number of events waiting in the application event queue.                        |
| `application-event-queue-time-avg`            | The average time an event spends in the application event queue, in<br/>milliseconds.       |
| `application-event-queue-time-max`            | The maximum time an event spends in the application event queue, in<br/>milliseconds.       |
| `application-event-queue-processing-time-avg` | The average time to process an event from the application event<br/>queue, in milliseconds. |
| `application-event-queue-processing-time-max` | The maximum time to process an event from the application event<br/>queue, in milliseconds. |
| `background-event-queue-size`                 | The current number of events waiting in the background event queue.                         |
| `background-event-queue-time-avg`             | The average time an event spends in the background event queue, in<br/>milliseconds.        |
| `background-event-queue-time-max`             | The maximum time an event spends in the background event queue, in<br/>milliseconds.        |
| `background-event-queue-processing-time-avg`  | The average time to process an event from the background event<br/>queue, in milliseconds.  |
| `background-event-queue-processing-time-max`  | The maximum time to process an event from the background event<br/>queue, in milliseconds.  |
| `unsent-requests-queue-size`                  | The current number of unsent requests in the queue.                                         |
| `unsent-requests-queue-time-avg`              | The average time a request spends in the unsent requests queue, in<br/>milliseconds.        |
| `unsent-requests-queue-time-max`              | The maximum time a request spends in the unsent requests queue, in<br/>milliseconds.        |

<a id="rust-client-producer-metrics"></a>

### Producer metric groups

The producer registers metrics in groups: `producer-metrics` for
client-level statistics and `producer-topic-metrics` for per-topic
statistics.

#### Producer metrics

The `producer-metrics` cover send throughput, batch characteristics,
request latency, throttling, and cumulative operation timing.

| Metric name                      | Description                                                                                                                               |
|----------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------|
| `batch-size-avg`                 | The average number of bytes sent per partition per request.                                                                               |
| `batch-size-max`                 | The maximum number of bytes sent per partition per request.                                                                               |
| `batch-split-rate`               | The average number of batch splits per second.                                                                                            |
| `batch-split-total`              | The total number of batch splits.                                                                                                         |
| `compression-rate-avg`           | The average compression rate of record batches, defined as the<br/>average ratio of the compressed batch size over the uncompressed size. |
| `flush-time-ns-total`            | Total time the producer has spent in `flush`, in nanoseconds.                                                                             |
| `metadata-age`                   | The age in seconds of the current producer metadata being used.                                                                           |
| `metadata-wait-time-ns-total`    | Total time the producer has spent waiting on topic metadata, in<br/>nanoseconds.                                                          |
| `produce-throttle-time-avg`      | The average time that a broker throttled a request, in milliseconds.                                                                      |
| `produce-throttle-time-max`      | The maximum time that a broker throttled a request, in milliseconds.                                                                      |
| `record-error-rate`              | The average per-second number of record sends that resulted in errors.                                                                    |
| `record-error-total`             | The total number of record sends that resulted in errors.                                                                                 |
| `record-queue-time-avg`          | The average time in milliseconds record batches spent in the send buffer.                                                                 |
| `record-queue-time-max`          | The maximum time in milliseconds record batches spent in the send buffer.                                                                 |
| `record-retry-rate`              | The average per-second number of retried record sends.                                                                                    |
| `record-retry-total`             | The total number of retried record sends.                                                                                                 |
| `record-send-rate`               | The average number of records sent per second.                                                                                            |
| `record-send-total`              | The total number of records sent.                                                                                                         |
| `record-size-avg`                | The average record size, in bytes.                                                                                                        |
| `record-size-max`                | The maximum record size, in bytes.                                                                                                        |
| `records-per-request-avg`        | The average number of records per request.                                                                                                |
| `request-latency-avg`            | The average request latency in milliseconds.                                                                                              |
| `request-latency-max`            | The maximum request latency in milliseconds.                                                                                              |
| `requests-in-flight`             | The current number of in-flight requests awaiting a response.                                                                             |
| `txn-abort-time-ns-total`        | Total time the producer has spent in `abort_transaction`, in<br/>nanoseconds.                                                             |
| `txn-begin-time-ns-total`        | Total time the producer has spent in `begin_transaction`, in<br/>nanoseconds.                                                             |
| `txn-commit-time-ns-total`       | Total time the producer has spent in `commit_transaction`, in<br/>nanoseconds.                                                            |
| `txn-init-time-ns-total`         | Total time the producer has spent in `init_transactions`, in<br/>nanoseconds.                                                             |
| `txn-send-offsets-time-ns-total` | Total time the producer has spent in `send_offsets_to_transaction`,<br/>in nanoseconds.                                                   |

#### Producer topic metrics

The `producer-topic-metrics` are scoped to individual topics and
carry a `topic` tag in addition to the `client-id` tag. They report
the same metric names regardless of topic, distinguished by their tags.

| Metric name          | Description                                                                                                                                               |
|----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------|
| `byte-rate`          | The average number of bytes sent per second for a topic.                                                                                                  |
| `byte-total`         | The total number of bytes sent for a topic.                                                                                                               |
| `compression-rate`   | The average compression rate of record batches for a topic, defined<br/>as the average ratio of the compressed batch size over the<br/>uncompressed size. |
| `record-error-rate`  | The average per-second number of record sends that resulted in errors<br/>for a topic.                                                                    |
| `record-error-total` | The total number of record sends that resulted in errors for a topic.                                                                                     |
| `record-retry-rate`  | The average per-second number of retried record sends for a topic.                                                                                        |
| `record-retry-total` | The total number of retried record sends for a topic.                                                                                                     |
| `record-send-rate`   | The average number of records sent per second for a topic.                                                                                                |
| `record-send-total`  | The total number of records sent for a topic.                                                                                                             |

<a id="rust-client-metrics-config"></a>

### Configure metrics collection

You can’t attach a metrics reporter to the producer, consumer, or admin
client. `metrics()` returns a read-only `MetricsView`, and the client
doesn’t support the `metric.reporters` property. If you set it, the client
ignores it and logs it at the `info` level as an unused configuration. To
export client metrics, read them periodically with `metrics()`, as described
in [Accessing metrics](#rust-client-accessing-metrics).

To control how the registry collects data, use the `metrics.recording.level`,
`metrics.num.samples`, and `metrics.sample.window.ms` configuration
properties. For full definitions and defaults, see
[Configuration for Rust Client for Kafka](configuration.md#rust-client-configuration).

The `MetricsReporter` trait works only with a `Metrics` registry that you
create yourself, for example to track your application’s own metrics with the
same framework. Implement the trait to receive callbacks when metrics are
registered, updated, or removed, and register it with
`Metrics::add_reporter()`:

```rust
use std::sync::Arc;

use confluent_kafka::common::Metric;
use confluent_kafka::common::metrics::{KafkaMetric, Metrics, MetricsReporter};

struct MyReporter;

impl MetricsReporter for MyReporter {
    fn init(&self, metrics: &[Arc<KafkaMetric>]) {
        println!("Initial metrics: {}", metrics.len());
    }

    fn metric_change(&self, metric: &Arc<KafkaMetric>) {
        println!("Metric updated: {}", metric.metric_name());
    }

    fn metric_removal(&self, metric: &Arc<KafkaMetric>) {
        println!("Metric removed: {}", metric.metric_name());
    }
}

let metrics = Metrics::new();
metrics.add_reporter(Arc::new(MyReporter));
```

`add_reporter()` calls `init()` with every metric already in the registry.
The trait also provides a `close()` method, which is called when the
metrics registry closes. Each method defaults to a no-operation
implementation. Override only the methods that you use.

<a id="rust-client-debugging"></a>

## Debugging

Debug the Rust client by enabling `debug`-level logging first, then inspect
metrics snapshots to identify the source of the problem.

### Start with logs

Enable `debug`-level logging for the client to see per-request output,
broker connection state, and internal event details not visible at the
`info` level:

```bash
RUST_LOG=confluent_kafka=debug cargo run
```

If you need even finer detail, including network I/O and individual fetch
handling, switch to `trace`:

```bash
RUST_LOG=confluent_kafka=trace cargo run
```

To keep client debug output while reducing noise from your own application,
set a separate level for each target. This example limits `my_app` to
warnings and errors and keeps the client at `debug`:

```bash
RUST_LOG=my_app=warn,confluent_kafka=debug cargo run
```

<a id="rust-client-debugging-errors"></a>

### Interpreting errors

The client’s `Error` type exposes a set of classification predicates that
let you decide how to handle an error without pattern-matching on every
variant:

| Method                        | When to use it                                                                                                                                                                                                                                               |
|-------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `error.is_retriable_error()`  | Returns `true` for transient errors, such as brief network interruptions<br/>and leader elections, that might succeed on a retry. The client retries<br/>these internally in most cases, but you can see them returned from<br/>`poll()` or `commit_sync()`. |
| `error.is_timeout_error()`    | Returns `true` when the operation exceeded its timeout. Check<br/>`request.timeout.ms` and the timeout passed to the call.                                                                                                                                   |
| `error.is_disconnect_error()` | Returns `true` when the broker connection was lost mid-request.<br/>Enable `debug` logging to see which broker was involved.                                                                                                                                 |
| `error.is_network_error()`    | Returns `true` for network-layer failures. Disconnects are reported<br/>separately by `is_disconnect_error()`.                                                                                                                                               |

Check retriability to decide whether to continue polling or shut down the consumer:

```rust
use std::time::Duration;

use confluent_kafka::consumer::Consumer;

loop {
    match consumer.poll(Duration::from_millis(500)).await {
        Ok(records) => { /* process records */ },
        Err(e) if e.is_retriable_error() => {
            // transient — wait briefly and retry the poll loop
            log::warn!("Retriable error, continuing: {e}");
        },
        Err(e) => {
            // non-retriable — log and shut down
            log::error!("Fatal consumer error: {e}");
            break;
        },
    }
}
```

<a id="rust-client-troubleshooting"></a>

### Troubleshooting

This section covers common issues and their resolutions. If you encounter
other issues, gather the diagnostic information in the
[Debugging checklist](#rust-client-debugging-checklist) and open a support case.

#### Consumer doesn’t start because of the group protocol

If the broker doesn’t support the consumer group protocol, the consumer hangs.
The Rust client supports only `group.protocol=consumer`, which must be explicitly
set. If `group.protocol=classic`, `KafkaConsumer::new` returns an
`UnsupportedVersion` error. If the broker doesn’t support KIP-848, the consumer
can stall silently while waiting for group coordinator responses.

If the broker doesn’t support the consumer group protocol, the consumer’s
heartbeat request fails, and the next call to `poll` returns an error for
which `is_unsupported_version_error()` returns `true`, with the message
`The cluster does not support the new CONSUMER group protocol`. The message
also suggests setting `group.protocol=classic`, which doesn’t apply to the
Rust client.

At the `error` level, the log shows a line that starts with
`ConsumerGroupHeartbeatRequest failed due to unsupported version`.

##### Fix the group protocol error

Verify that you’re using Kafka 4.0 or later, Confluent Platform 8.0 or later,
or Confluent Cloud. Confirm that `group.protocol=consumer` is in your configuration.

#### Rebalances are frequent or failing

Frequent rebalances degrade throughput and can cause repeated reprocessing
of records.

##### Metrics for rebalances

* `rebalance-rate-per-hour` and `failed-rebalance-rate-per-hour`: A
  spike in either indicates instability.
* `time-between-poll-max`: A high value means the application is spending
  too long between `poll()` calls.
* `last-heartbeat-seconds-ago`: A growing value means the consumer isn’t
  completing heartbeats, either because the event loop is blocked or because
  the coordinator isn’t responding.

##### Fix frequent rebalances

Check the following configuration values:

* If record processing takes longer than `max.poll.interval.ms`,
  the consumer leaves the group, which triggers a rebalance. Increase the
  `max.poll.interval.ms` value or process fewer records per poll batch
  using `max.poll.records`.
* With the KIP-848 protocol, the broker controls the session
  timeout with `group.consumer.session.timeout.ms`. The client-side
  `session.timeout.ms` property has no effect. If the consumer doesn’t send
  a heartbeat within the broker’s session timeout, the broker considers it
  failed and removes it from the group. Ensure the event loop isn’t blocked.

#### Consumer lag is growing

Rising consumer lag means producers are writing records faster than the consumer
can process them.

##### Metrics for consumer lag

* `records-lag-max`: The maximum lag across all partitions.
* `records-lag` tagged with `topic` and `partition`: The lag for each
  partition. Use it to identify hot partitions.

##### Reduce consumer lag

* Increase the number of consumer instances in the group, up to the number
  of partitions.
* If processing is fast and network round-trips are the bottleneck, increase
  `max.poll.records`.
* Profile the application code that runs between `poll()` calls.

#### Commit failures

Commit failures can cause records to be reprocessed after a restart or a
rebalance.

##### Metrics for commit failures

* `commit-latency-avg` and `commit-latency-max`: High values indicate
  broker pressure or network latency.
* `commit-rate`: An unexpected drop might indicate the commit loop is
  exiting early.

##### Fix commit failures

`commit_sync()` returns an error on failure. Check whether
`error.is_retriable_error()` returns `true`. If it does, retry the commit.

A `CoordinatorNotAvailable` error means the group coordinator is unavailable,
and a `NotCoordinator` error means that the broker contacted is no longer the
group coordinator, for example because the coordinator moved. These errors are
retriable, and the client refreshes coordinator metadata automatically.

#### High fetch latency

High fetch latency reduces throughput and increases end-to-end processing
delay.

##### Metrics for fetch latency

* `fetch-latency-avg` and `fetch-latency-max`: High values
  indicate network delay, broker-side slowness, or fetch configuration issues.
* `fetch-throttle-time-avg` and `fetch-throttle-time-max`: A nonzero value
  means the broker is rate-limiting the consumer, which is distinct from
  organic latency. If `fetch-throttle-time-avg` accounts for most of
  `fetch-latency-avg`, the root cause is quota exhaustion, not a
  performance problem.
* `bytes-consumed-rate`: A low rate combined with high latency points to
  small, infrequent fetches.

##### Reduce fetch latency

* Check broker-side metrics for disk I/O or CPU pressure.
* Review `fetch.max.wait.ms` and `fetch.min.bytes`: increasing
  `fetch.min.bytes` batches more data per request at the cost of added
  latency.
* Review `fetch-throttle-time-avg`: a nonzero value means the broker is
  rate-limiting the consumer.

#### Internal event queues backing up

The consumer uses asynchronous event queues between the application thread and
the background network thread. Queues that grow without draining indicate a
bottleneck inside the client’s internal processing loop.

##### Metrics for event queue backlogs

* `application-event-queue-size` and `background-event-queue-size`: If
  either is consistently nonzero, the corresponding processing thread is
  not keeping up.
* `unsent-requests-queue-size`: A growing queue means the network thread
  can’t send requests fast enough. Check for broker connectivity issues or
  high broker latency.
* `time-between-network-thread-poll-max`: A spike here indicates that the
  network thread was blocked, which can delay heartbeats and trigger a
  rebalance.

##### Clear event queue backlogs

Check that application code between `poll()` calls isn’t
blocking the Tokio runtime. Check broker connectivity. If queues stay backed
up, collect the information in [Debugging checklist](#rust-client-debugging-checklist) and open a
support case.

<a id="rust-client-debugging-checklist"></a>

### Debugging checklist

When opening a support case or filing a bug report, collect the following
information:

* The full error message and, if available, the error variant name from
  `format!("{:?}", error)`
* Log output at `debug` level from the moment the problem starts, using
  `RUST_LOG=confluent_kafka=debug`
* A snapshot of the metrics returned by `consumer.metrics()` at the time
  of the problem, printed as `(name, value)` pairs
* The consumer configuration, without credentials: `bootstrap.servers`,
  `group.id`, `group.protocol`, timeout settings, and any non-default
  values
* The Kafka broker version or Confluent Cloud cluster type
* The Rust client version from `Cargo.lock`