[feature request] Expose librdkafka client statistics as OpenTelemetry metrics

[tombiddulph]

Component

OpenTelemetry.Instrumentation.ConfluentKafka

Is your feature request related to a problem?

The instrumentation currently emits these messaging operation metrics:

• messaging.client.operation.duration
• messaging.client.sent.messages
• messaging.client.consumed.messages

These describe application-level send and consume operations, but do not expose the operational state of the underlying Kafka client. Important signals such as consumer lag, producer queue depth, retries, request latency, throttling, and bytes transferred are therefore unavailable.

The OpenTelemetry Java Kafka instrumentation exposes a broad set of native Kafka client metrics through Kafka's MetricsReporter interface. Confluent.Kafka does not provide the same interface, but its underlying librdkafka client emits detailed statistics as JSON through SetStatisticsHandler when statistics.interval.ms is configured.

What is the expected behavior?

Provide an opt-in bridge that converts a curated set of librdkafka statistics into OpenTelemetry metrics.

An initial set could include:

Consumer

• Records and bytes consumed
• Consumer lag by topic and partition
• Assigned partition count
• Rebalance count
• Fetch queue messages and bytes

Producer

• Records and bytes transmitted
• Queued messages and bytes
• Messages or requests in flight
• Request retries
• Request timeouts

Shared client and broker

• Requests and responses
• Bytes transmitted and received
• Request latency
• Broker throttle time
• Connection state or count

The exact metric names, units, instrument types, and attributes should be agreed upon before implementation. The goal would be equivalent operational coverage to the Java instrumentation where the librdkafka measurements have compatible semantics, rather than strict one-to-one parity.

Possible implementation

The instrumentation could:

1. Respect an existing statistics.interval.ms value or configure a user-specified interval.
2. Install a composite statistics handler that preserves an existing user SetStatisticsHandler callback.
3. Parse the periodic librdkafka statistics JSON.
4. Store the latest snapshot for each producer or consumer.
5. Publish selected values using observable counters and gauges.
6. Remove observations when the corresponding client is disposed.

Configuration might resemble:

metering.AddKafkaConsumerInstrumentation<string, string>(
    consumerBuilder,
    options =>
    {
        options.EnableClientMetrics = true;
        options.StatisticsInterval = TimeSpan.FromSeconds(10);
        options.EnablePartitionMetrics = true;
    });

This API is only illustrative.

Cardinality and performance considerations

Native statistics collection should be opt-in because generating and parsing the JSON has a runtime cost.

Low-cardinality client-level metrics could be enabled separately from topic-, partition-, and broker-level metrics. Partition-level consumer lag is particularly valuable, but could create a large number of time series.

The implementation should also:

• Avoid exporting librdkafka's generated handle name as an attribute.
• Avoid double-counting values available at client, broker, topic, and partition granularities.
• Distinguish existing application operation counters from native transmitted or received message counters.
• Ignore unknown JSON fields and treat missing fields as unavailable rather than zero.
• Handle counter resets when clients restart.
• Filter librdkafka sentinel offset and lag values.
• Ensure parsing or instrumentation failures cannot interrupt Kafka processing.

Which alternative solutions or features have you considered?

Applications can currently register their own SetStatisticsHandler, parse the JSON, and implement custom metrics. However, this duplicates mapping, lifecycle, cardinality, and callback-composition logic across applications.

A separate package could provide this bridge, but it would need to duplicate or integrate with the producer and consumer wrapping already performed by this instrumentation.

Additional context

Relevant references:

• Current Confluent.Kafka metrics
• OpenTelemetry Java Kafka client metric mappings
• librdkafka statistics reference
• librdkafka configuration reference
• Related semantic-convention work: [feature request] Kafka instrumentation should follow the Semantic Conventions #2567
• Related cross-language parity work: ConfluentKafka instrumentation: add messaging.kafka.cluster.id attribute #4738

I would be happy to produce a proof of concept for an agreed initial metric set and API design.

[matt-hensley]

Implementing a reasonable subset of metrics makes sense. Java instrumentation already defined 174(!) metrics. Likely far more than we need.

Given messaging has been stalled since 2024, following Java's lead with a set of core metrics makes sense IMO.

[tombiddulph]

I've put together a working PoC based on the feedback here:

• PoC branch
• Implementation commit
• Design and implementation plan

The PoC currently includes:

• Opt-in EnableClientMetrics, StatisticsInterval, and EnablePartitionMetrics options for producers and consumers.
• Preservation of explicitly configured statistics.interval.ms.
• A composite statistics handler that preserves the application's existing callback and prevents instrumentation failures from reaching librdkafka.
• Name-based, tolerant JSON parsing with missing fields treated as unavailable.
• A curated set of 29 observable producer, consumer, broker, and partition instruments following the Java kafka-clients naming and attributes.
• Opt-in partition-level consumer lag.
• Per-source monotonic counter aggregation, including clients sharing the same client.id.
• Removal of observations when clients are disposed.
• EventSource diagnostics for parser failures and the detected librdkafka version.
• Versioned librdkafka fixtures, schema conformance tests, unit tests, and a Docker-gated integration test.
• README and public API documentation.

The implementation builds cleanly for net10.0, net8.0, netstandard2.0, and net462. Unit tests pass on net10.0, net8.0, and net462; the Kafka integration test is included but was not runnable locally because Docker was unavailable.

Known mapping differences from Java are documented:

• rebalance_total counts librdkafka assignment and revocation events.
• record_retry_total is client-level because librdkafka exposes request retries per broker rather than record retries per topic.

The 10-second default interval, explicit configuration precedence, partition gating, metric names, and mappings are all open to maintainer feedback before turning this into a draft PR.