Emit messaging operation metrics (publish/receive duration) from Kafka instrumentation by maryantocinn · Pull Request #19107 · open-telemetry/opentelemetry-java-instrumentation

maryantocinn · 2026-06-30T09:36:28Z

Summary

Emits the semantic-convention messaging metrics from the Kafka instrumentation by registering the existing operation-metrics listeners on the producer and consumer instrumenters in KafkaInstrumenterFactory (kafka-clients-common-0.11). This mirrors the Pulsar instrumentation, which already registers MessagingProducerMetrics / MessagingConsumerMetrics.

Resolves #19094.

What changed

Added .addOperationMetrics(...) to the producer and consumer receive instrumenters in KafkaInstrumenterFactory:

Producer → MessagingProducerMetrics.get()
Consumer receive → MessagingConsumerMetrics.get() (gated by messagingReceiveInstrumentationEnabled via the builder's setEnabled(...))

The consumer process instrumenter is intentionally left untouched, matching Pulsar: MessagingConsumerMetrics emits messaging.receive.*, which is the wrong metric to record for a process operation (and there is no messaging.process.* listener in instrumentation-api-incubator).

import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingConsumerMetrics;
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingProducerMetrics;
// ...
// producer
    .addOperationMetrics(MessagingProducerMetrics.get())
// consumer receive
    .addOperationMetrics(MessagingConsumerMetrics.get())

Metrics added

Instrumenter	Listener	Metric(s)
producer	`MessagingProducerMetrics`	`messaging.publish.duration` (histogram, s)
consumer receive	`MessagingConsumerMetrics`	`messaging.receive.duration` (histogram, s), `messaging.receive.messages` (counter)

Scope / impact

A single shared factory (kafka-clients-common-0.11) is used by the javaagent Kafka instrumentation, the kafka-clients-2.6 library (KafkaTelemetry, Spring Boot starter, manual users), and spring-kafka — so all paths get the metrics from one change.

Design note for reviewers

Two points to confirm:

Experimental default-on vs flag. The messaging-metrics semconv is still experimental. This follows the Pulsar precedent of emitting these without a dedicated flag; happy to gate behind an experimental toggle instead if you prefer.
Consumer metrics off by default. Because Kafka's receive instrumenter is gated by otel.instrumentation.messaging.experimental.receive-telemetry.enabled (off by default), messaging.receive.duration / messaging.receive.messages only emit when that flag is enabled. The listener is deliberately not added to the process instrumenter, since MessagingConsumerMetrics records messaging.receive.* and that would be the wrong metric for a process operation. Let me know if you'd prefer the consumer metrics surfaced by default.

Test plan

Unit/integration tests assert messaging.publish.duration (producer) and messaging.receive.duration / messaging.receive.messages (receive, with receive telemetry enabled) are emitted with the expected attributes
./gradlew spotlessApply clean
Affected :instrumentation:kafka:... module tests pass

…a instrumentation

linux-foundation-easycla · 2026-06-30T09:36:36Z

The committers listed above are authorized under a signed CLA.

✅ login: maryantocinn / name: Maryanto (0208420)

Copilot

Pull request overview

This PR makes the Kafka instrumentation emit the semantic-convention messaging operation metrics by registering the existing operation-metrics listeners in the shared KafkaInstrumenterFactory (kafka-clients-common-0.11). Because this factory is shared, the change covers the javaagent Kafka instrumentation, the kafka-clients-2.6 library (KafkaTelemetry, Spring Boot starter, manual users), and spring-kafka. It mirrors the existing Pulsar instrumentation, resolving #19094.

Changes:

Register MessagingProducerMetrics.get() on the producer instrumenter (emits messaging.publish.duration).
Register MessagingConsumerMetrics.get() on the consumer receive instrumenter (emits messaging.receive.duration / messaging.receive.messages, gated by messagingReceiveInstrumentationEnabled); the process instrumenter is intentionally left untouched.
Add metric assertions in the wrapper tests, with isReceiveTelemetryEnabled() controlling whether the receive metrics are asserted.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
`kafka-clients-common-0.11/.../KafkaInstrumenterFactory.java`	Adds `addOperationMetrics(...)` to the producer and consumer-receive instrumenters and the corresponding imports.
`kafka-clients-2.6/.../AbstractWrapperTest.java`	Adds `assertMessagingMetrics()` verifying the publish/receive metric name/unit/description; switches `assertThat` import to `OpenTelemetryAssertions`.
`kafka-clients-2.6/.../WrapperTest.java`	Overrides `isReceiveTelemetryEnabled()` to `true` and removes the now-redundant assertj `assertThat` import.

laurit · 2026-07-02T09:00:38Z

I think we should add these for all messaging instrumentation (doing just kafka in this PR is fine, the rest can be handled separately). It is not ideal that we only have messaging.receive.duration and messaging.publish.duration but nothing for process. I think we have the following choices

leave it as is, don't add anything for process
add the missing process metrics based on https://github.com/open-telemetry/semantic-conventions/blob/v1.24.0/docs/messaging/messaging-metrics.md (there the metric names contains deliver instead of process)
switch to metrics from current semconv https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-metrics.md Since our messaging implementation doesn't really follow any version of the semantic conventions this might not be that bad. Since the messaging metrics are currently only implemented for pulsar the backwards compatibility concerns are limited.

@trask do you have any preference on how we should handle this?

maryantocinn · 2026-07-06T07:37:05Z

@laurit Thanks for the review! I'd go with the third option. One extra reason from my side: we run both Java and Go services in my org and we want the same messaging metrics from both, so following the current semconv names (messaging.client.operation.duration etc) means one set of dashboards/alerts for everything.

I'm willing to implement it. This is roughly what I have in mind:

MessagingProducerMetrics emits messaging.client.operation.duration and messaging.client.sent.messages
MessagingConsumerMetrics emits messaging.client.operation.duration and messaging.client.consumed.messages
add a new MessagingProcessMetrics for messaging.process.duration and register it on the kafka process/batch process instrumenters, so process is covered by default
consumed.messages only counted on receive, since the spec says it should be reported once per delivery and not again on process
pulsar would get the new names too since the listeners are shared, so I'd update its tests as well (and its process instrumenter could also record messaging.process.duration)

I wouldn't touch the attributes for now, they'd stay the old ones (messaging.operation etc) - migrating those is more of a span conventions thing, probably better done separately.

If that sounds good I'll update the PR.

trask · 2026-07-14T21:30:36Z

@trask do you have any preference on how we should handle this?

It has downsides, but I'd lean towards continuing to follow https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/README.md, which basically says, emit the old semconv by default, and only emit new semconv under OTEL_SEMCONV_STABILITY_OPT_IN (and in our case starting with 3.0, OTEL_SEMCONV_STABILITY_PREVIEW).

We could bump to latest (even before stable), but ideally only after we have implemented the same conventions consistently across all of our messaging instrumentations. It would probably be confusing for some of our messaging instrumentations to emit earlier semconv and some later semconv by default.

laurit · 2026-07-15T14:24:27Z

It has downsides, but I'd lean towards continuing to follow https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/README.md, which basically says, emit the old semconv by default, and only emit new semconv under OTEL_SEMCONV_STABILITY_OPT_IN (and in our case starting with 3.0, OTEL_SEMCONV_STABILITY_PREVIEW).

We could bump to latest (even before stable), but ideally only after we have implemented the same conventions consistently across all of our messaging instrumentations. It would probably be confusing for some of our messaging instrumentations to emit earlier semconv and some later semconv by default.

The thing is that our messaging semconv don't really follow a particular version of the semconv. The version that is closest to how we emit spans doesn't define metrics at all. The messaging metrics are currently only enabled for pulsar so I'd say we can probably allow incompatible changes if we so desire since their usage is limited.
@trask do I understand correctly that your preference is to use https://github.com/open-telemetry/semantic-conventions/blob/v1.25.0/docs/messaging/messaging-metrics.md for metrics (1.24.0 has deliver instead of process)?

trask · 2026-07-16T04:30:24Z

The thing is that our messaging semconv don't really follow a particular version of the semconv.

You've convinced me 😅.

But if we're going to do it, let's take the opportunity to sync them all up for 3.0:

#19233

opentelemetry-pr-dashboard · 2026-07-18T13:02:10Z

Pull request dashboard status

Status last refreshed: 2026-07-26 21:34:24 UTC.

Waiting on: Reviewers
Next step: Review the latest changes.

This automated status or its linked feedback items may be incorrect. If something looks wrong, please report it with the result you expected.

maryantocinn · 2026-07-20T04:35:39Z

@trask Sounds good from my end, agreed on aligning to v1.43 for 3.0. Once it's merged, I'll rebase and re-scope this PR to just the Kafka instrumentation on top of the shared contract.

Happy to take that follow-up.

Emit messaging operation metrics (publish/receive duration) from Kafk…

0208420

…a instrumentation

Copilot AI review requested due to automatic review settings June 30, 2026 09:36

maryantocinn requested a review from a team as a code owner June 30, 2026 09:36

Copilot started reviewing on behalf of maryantocinn June 30, 2026 09:36 View session

opentelemetry-pr-dashboard Bot mentioned this pull request Jun 30, 2026

Pull Request Dashboard #18435

Open

Merge branch 'main' into add-semconv-metrics-kafka

1193528

Copilot AI reviewed Jun 30, 2026

View reviewed changes

Merge branch 'main' into add-semconv-metrics-kafka

df94f54

jaydeluca mentioned this pull request Jun 30, 2026

Add messaging metrics to Kafka client instrumentation #19110

Closed

trask mentioned this pull request Jul 18, 2026

PR Dashboard: Investigate incorrect PR statuses open-telemetry/shared-workflows#106

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Emit messaging operation metrics (publish/receive duration) from Kafka instrumentation#19107

Emit messaging operation metrics (publish/receive duration) from Kafka instrumentation#19107
maryantocinn wants to merge 3 commits into
open-telemetry:mainfrom
maryantocinn:add-semconv-metrics-kafka

maryantocinn commented Jun 30, 2026

Uh oh!

linux-foundation-easycla Bot commented Jun 30, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

laurit commented Jul 2, 2026 •

edited

Loading

Uh oh!

maryantocinn commented Jul 6, 2026 •

edited

Loading

Uh oh!

trask commented Jul 14, 2026

Uh oh!

laurit commented Jul 15, 2026

Uh oh!

trask commented Jul 16, 2026

Uh oh!

opentelemetry-pr-dashboard Bot commented Jul 18, 2026 •

edited

Loading

Uh oh!

maryantocinn commented Jul 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

maryantocinn commented Jun 30, 2026

Summary

What changed

Metrics added

Scope / impact

Design note for reviewers

Test plan

Uh oh!

linux-foundation-easycla Bot commented Jun 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

laurit commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

maryantocinn commented Jul 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

trask commented Jul 14, 2026

Uh oh!

laurit commented Jul 15, 2026

Uh oh!

trask commented Jul 16, 2026

Uh oh!

opentelemetry-pr-dashboard Bot commented Jul 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Pull request dashboard status

Uh oh!

maryantocinn commented Jul 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

linux-foundation-easycla Bot commented Jun 30, 2026 •

edited

Loading

laurit commented Jul 2, 2026 •

edited

Loading

maryantocinn commented Jul 6, 2026 •

edited

Loading

opentelemetry-pr-dashboard Bot commented Jul 18, 2026 •

edited

Loading