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.

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 behavior.
`info`	High-level lifecycle events such as client creation and shutdown.
`debug`	Per-request activity, broker connections, and internal state changes useful for diagnosing problems.
`trace`	Fine-grained detail including network I/O and individual record handling. Produces a large volume of output.

Set up logging

To set up logging, add the log crate, log = "0.4", to your project. Set up a concrete logging backend to control how and where the output appears.

env_logger is a common choice for command-line applications and examples.

Add it to your project:
```
cargo add env_logger
```

Initialize env_logger at the start of your application, before creating your client:

fn main() {
    env_logger::init();
    // create your producer, consumer, or admin client here
}

Before running your application, set the RUST_LOG environment variable to control which log records appear:
```
RUST_LOG=confluent_kafka=debug cargo run
```
Alternatively, set the level for only the client crate:
```
RUST_LOG=confluent_kafka=info cargo run
```
You can also combine the log level with your application’s own logging:
```
RUST_LOG=my_app=debug,confluent_kafka=warn cargo run
```

For the full syntax, see the env_logger documentation.

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

Accessing metrics

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

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().

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

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.

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 `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 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 the current window. Lag is based on the current position, not the committed offset.
`records-lead-min`	The minimum lead, in number of records, for any partition in 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 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 callback.
`partition-assigned-latency-max`	The maximum time taken for a partition-assigned rebalance listener callback.
`partition-revoked-latency-avg`	The average time taken for a partition-revoked rebalance listener callback.
`partition-revoked-latency-max`	The maximum time taken for a partition-revoked rebalance listener callback.
`partition-lost-latency-avg`	The average time taken for a partition-lost rebalance listener callback.
`partition-lost-latency-max`	The maximum time taken for a partition-lost rebalance listener 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 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 milliseconds.
`time-between-network-thread-poll-max`	The maximum time between network thread poll iterations, in 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 milliseconds.
`application-event-queue-time-max`	The maximum time an event spends in the application event queue, in milliseconds.
`application-event-queue-processing-time-avg`	The average time to process an event from the application event queue, in milliseconds.
`application-event-queue-processing-time-max`	The maximum time to process an event from the application event 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 milliseconds.
`background-event-queue-time-max`	The maximum time an event spends in the background event queue, in milliseconds.
`background-event-queue-processing-time-avg`	The average time to process an event from the background event queue, in milliseconds.
`background-event-queue-processing-time-max`	The maximum time to process an event from the background event 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 milliseconds.
`unsent-requests-queue-time-max`	The maximum time a request spends in the unsent requests queue, in milliseconds.

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 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 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 nanoseconds.
`txn-begin-time-ns-total`	Total time the producer has spent in `begin_transaction`, in nanoseconds.
`txn-commit-time-ns-total`	Total time the producer has spent in `commit_transaction`, in nanoseconds.
`txn-init-time-ns-total`	Total time the producer has spent in `init_transactions`, in nanoseconds.
`txn-send-offsets-time-ns-total`	Total time the producer has spent in `send_offsets_to_transaction`, 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 as the average ratio of the compressed batch size over the uncompressed size.
`record-error-rate`	The average per-second number of record sends that resulted in errors 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.

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.

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.

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():

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.

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:

RUST_LOG=confluent_kafka=debug cargo run

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

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:

RUST_LOG=my_app=warn,confluent_kafka=debug cargo run

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

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

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;
        },
    }
}

Troubleshooting

This section covers common issues and their resolutions. If you encounter other issues, gather the diagnostic information in the 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 and open a support case.

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