You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2026/08/28 02:12:52 UTC
(pulsar) branch branch-4.2 updated: [fix][client] Fix incorrect topic in OTel spans for multi-topic consumers
This is an automated email from the ASF dual-hosted git repository.
Technoboy- pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new be1082174535 [fix][client] Fix incorrect topic in OTel spans for multi-topic consumers
be1082174535 is described below
commit be1082174535c3a6d57637e35df3578f6a652cca
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Fri Aug 28 10:10:32 2026 +0800
[fix][client] Fix incorrect topic in OTel spans for multi-topic consumers
Use the actual message topic when creating OpenTelemetry consumer spans for multi-topic consumers, and prefer the TopicMessageId owner topic on branch-4.2 where TopicMessageImpl#getTopicName can still expose the synthetic MultiTopicsConsumer name.
(cherry picked from commit 0c37fd30df8b0707490e15165a2176e8ea8e25d6)
---
.../OpenTelemetryTracingIntegrationTest.java | 17 +++++++++---
.../tracing/OpenTelemetryConsumerInterceptor.java | 31 +++++++++++++++++-----
2 files changed, 37 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java
index fcc0bd1776eb..3b3c33a47791 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java
@@ -363,11 +363,20 @@ public class OpenTelemetryTracingIntegrationTest extends BrokerTestBase {
// Verify spans for both topics
List<SpanData> spans = spanExporter.getFinishedSpanItems();
- long consumerSpans = spans.stream()
+ List<SpanData> consumerSpans = spans.stream()
.filter(s -> s.getKind() == SpanKind.CONSUMER)
- .count();
-
- assertEquals(consumerSpans, 2);
+ .collect(java.util.stream.Collectors.toList());
+ Set<String> consumerSpanTopics = consumerSpans.stream()
+ .map(s -> s.getAttributes().get(
+ io.opentelemetry.api.common.AttributeKey.stringKey("messaging.destination.name")))
+ .collect(java.util.stream.Collectors.toSet());
+ Set<String> consumerSpanNames = consumerSpans.stream()
+ .map(SpanData::getName)
+ .collect(java.util.stream.Collectors.toSet());
+
+ assertEquals(consumerSpans.size(), 2);
+ assertEquals(consumerSpanTopics, Set.of(topic1, topic2));
+ assertEquals(consumerSpanNames, Set.of("process " + topic1, "process " + topic2));
producer1.close();
producer2.close();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryConsumerInterceptor.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryConsumerInterceptor.java
index eece10b76ff9..4dacbdd88742 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryConsumerInterceptor.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryConsumerInterceptor.java
@@ -68,7 +68,11 @@ public class OpenTelemetryConsumerInterceptor<T> implements ConsumerInterceptor<
private Tracer tracer;
private TextMapPropagator propagator;
- private String topic;
+ /**
+ * Topic returned by the consumer. Used as a fallback when a message has no topic name,
+ * and as the topic key for acknowledgment callbacks that only carry a MessageId.
+ */
+ private String consumerTopic;
private String subscription;
private boolean initialized = false;
@@ -96,7 +100,19 @@ public class OpenTelemetryConsumerInterceptor<T> implements ConsumerInterceptor<
if (messageId instanceof TopicMessageId) {
return ((TopicMessageId) messageId).getOwnerTopic();
}
- return topic != null ? topic : "";
+ return consumerTopic != null ? consumerTopic : "";
+ }
+
+ private String getMessageTopic(Message<T> message) {
+ MessageId messageId = message.getMessageId();
+ if (messageId instanceof TopicMessageId) {
+ return ((TopicMessageId) messageId).getOwnerTopic();
+ }
+ String messageTopic = message.getTopicName();
+ if (messageTopic != null && !messageTopic.isEmpty()) {
+ return messageTopic;
+ }
+ return consumerTopic != null ? consumerTopic : "";
}
/**
@@ -150,15 +166,16 @@ public class OpenTelemetryConsumerInterceptor<T> implements ConsumerInterceptor<
}
try {
- if (topic == null) {
- topic = consumer.getTopic();
+ if (consumerTopic == null) {
+ consumerTopic = consumer.getTopic();
}
if (subscription == null) {
subscription = consumer.getSubscription();
}
+ String messageTopic = getMessageTopic(message);
// Create a consumer span for this message
- Span span = TracingContext.createConsumerSpan(tracer, topic, subscription, message, propagator);
+ Span span = TracingContext.createConsumerSpan(tracer, messageTopic, subscription, message, propagator);
if (TracingContext.isValid(span)) {
MessageId messageId = message.getMessageId();
@@ -175,7 +192,7 @@ public class OpenTelemetryConsumerInterceptor<T> implements ConsumerInterceptor<
((TraceableMessageId) messageId).setTracingSpan(span);
}
- log.debug("Created consumer span for message {} on topic {}", messageId, topic);
+ log.debug("Created consumer span for message {} on topic {}", messageId, messageTopic);
}
} catch (Exception e) {
log.error("Error creating consumer span", e);
@@ -370,4 +387,4 @@ public class OpenTelemetryConsumerInterceptor<T> implements ConsumerInterceptor<
}
}
}
-}
\ No newline at end of file
+}