Kafka instrumentation does not emit semconv messaging metrics (messaging.publish.duration / messaging.receive.duration)

Is your feature request related to a problem? Please describe.

The Kafka client instrumentation emits producer/consumer spans but does not emit the semantic-convention messaging metrics messaging.publish.duration and messaging.receive.duration (+messaging.receive.messages).

The shared factory KafkaInstrumenterFactory (module instrumentation/kafka/kafka-clients/kafka-clients-common-0.11) never calls addOperationMetrics(...) on its producer/consumer instrumenters, so these metrics are never produced — and there is no configuration flag to turn them on today.

Note this is unrelated to the Kafka client native metrics (kafka.producer.* / kafka.consumer.*) bridged via OpenTelemetryMetricsReporter; those are the client's own metrics, not the portable semconv messaging metrics.

Describe the solution you'd like

Register the existing operation-metrics listeners on the Kafka instrumenters, mirroring the Pulsar instrumentation which already does this:

• Producer instrumenter → MessagingProducerMetrics.get() → emits messaging.publish.duration
• Consumer process/receive instrumenters → MessagingConsumerMetrics.get() → emits messaging.receive.duration and messaging.receive.messages

Both listeners already exist in instrumentation-api-incubator (io.opentelemetry.instrumentation.api.incubator.semconv.messaging.*). Because this is the shared factory, one change covers the javaagent, the kafka-clients-2.6 library (and therefore KafkaTelemetry / the Spring Boot starter), and spring-kafka.

Open question for maintainers: the messaging-metrics semconv is still experimental. Pulsar emits these unconditionally (precedent for default-on), but you may prefer gating them behind an experimental flag. Happy to follow whichever direction you prefer.

Describe alternatives you've considered

• Do nothing / rely on native Kafka client metrics (kafka.producer.*): rejected — those are not the portable semconv messaging metrics and differ per messaging system.
• Wire metrics in the Spring Boot starter / KafkaTelemetry: rejected — those only wire up the existing instrumentation. The metric set is decided in KafkaInstrumenterFactory, so the shared factory is the correct, high-leverage location.

Additional context

Precedent (reference implementation): instrumentation/pulsar/pulsar-2.8/javaagent/.../telemetry/PulsarSingletons.java already registers MessagingProducerMetrics / MessagingConsumerMetrics.

Semconv: https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-metrics.md

Verified the producer/consumer instrumenters in KafkaInstrumenterFactory have no addOperationMetrics(...) on main. Other span-only messaging instrumentations (JMS, SQS, RocketMQ) could get the same treatment, but this request is scoped to Kafka. I'm happy to open a PR.

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}

Start in instrumentation/kafka/kafka-clients/kafka-clients-common-0.11 at KafkaInstrumenterFactory, then compare the operation-metrics registration in the referenced PulsarSingletons.java implementation. Trace the producer and consumer instrumenters and existing MessagingProducerMetrics/MessagingConsumerMetrics listeners. Done means Kafka emits the requested publish and receive semantic-convention metrics, with the default-on or experimental-flag behavior resolved by maintainers.