You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Technoboy- (via GitHub)" <gi...@apache.org> on 2025/10/18 02:19:27 UTC
[PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Technoboy- opened a new pull request, #24873:
URL: https://github.com/apache/pulsar/pull/24873
### Motivation
Implementation of [PIP-446](https://github.com/apache/pulsar/pull/24857)
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2666959987
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/TracingContext.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Utility class for managing OpenTelemetry tracing context in Pulsar messages.
+ */
+public class TracingContext {
+
+ private static final TextMapGetter<Map<String, String>> GETTER = new TextMapGetter<Map<String, String>>() {
+ @Override
+ public Iterable<String> keys(Map<String, String> carrier) {
+ return carrier.keySet();
+ }
+
+ @Nullable
+ @Override
+ public String get(@Nullable Map<String, String> carrier, String key) {
+ return carrier != null ? carrier.get(key) : null;
+ }
+ };
+
+ private static final TextMapSetter<Map<String, String>> SETTER = (carrier, key, value) -> {
+ if (carrier != null) {
+ carrier.put(key, value);
+ }
+ };
+
+ /**
+ * Extract trace context from message properties.
+ *
+ * @param message the message to extract context from
+ * @param propagator the text map propagator to use
+ * @return the extracted context
+ */
+ public static Context extractContext(Message<?> message, TextMapPropagator propagator) {
+ if (message == null || propagator == null) {
+ return Context.current();
+ }
+ return propagator.extract(Context.current(), message.getProperties(), GETTER);
+ }
+
+ /**
+ * Inject trace context into message properties.
+ *
+ * @param messageBuilder the message builder to inject context into
+ * @param context the context to inject
+ * @param propagator the text map propagator to use
+ */
+ public static <T> void injectContext(TypedMessageBuilder<T> messageBuilder, Context context,
+ TextMapPropagator propagator) {
+ if (messageBuilder == null || context == null || propagator == null) {
+ return;
+ }
+
+ Map<String, String> carrier = new HashMap<>();
+ propagator.inject(context, carrier, SETTER);
+
+ for (Map.Entry<String, String> entry : carrier.entrySet()) {
+ messageBuilder.property(entry.getKey(), entry.getValue());
+ }
+ }
+
+ /**
+ * Create a producer span for message publishing.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param parentContext the parent context (may be null)
+ * @return the created span
+ */
+ public static Span createProducerSpan(Tracer tracer, String topic, @Nullable Context parentContext) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context context = parentContext != null ? parentContext : Context.current();
+ return tracer.spanBuilder("send " + topic)
+ .setParent(context)
+ .setSpanKind(SpanKind.PRODUCER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.operation.name", "send")
+ .startSpan();
+ }
+
+ /**
+ * Create a consumer span for message consumption.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param subscription the subscription name
+ * @param message the message being consumed
+ * @param propagator the text map propagator to use for context extraction
+ * @return the created span
+ */
+ public static Span createConsumerSpan(Tracer tracer, String topic, String subscription, Message<?> message,
+ TextMapPropagator propagator) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context parentContext = extractContext(message, propagator);
+
+ return tracer.spanBuilder("process " + topic)
+ .setParent(parentContext)
+ .setSpanKind(SpanKind.CONSUMER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.destination.subscription.name", subscription)
+ .setAttribute("messaging.operation.name", "process")
+ .setAttribute("messaging.message.id", message.getMessageId().toString())
+ .startSpan();
+ }
+
+ /**
+ * Mark a span as successful and end it.
+ *
+ * @param span the span to end
+ */
+ public static void endSpan(Span span) {
+ if (span != null && span.isRecording()) {
+ span.setStatus(StatusCode.OK);
+ span.end();
+ }
+ }
+
+ /**
+ * Mark a span as failed with an exception and end it.
+ *
+ * @param span the span to end
+ * @param throwable the exception that caused the failure
+ */
+ public static void endSpan(Span span, Throwable throwable) {
+ if (span != null && span.isRecording()) {
+ span.setStatus(StatusCode.ERROR, throwable.getMessage());
+ span.recordException(throwable);
+ span.end();
+ }
Review Comment:
yes, right. updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2623635765
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
+ }
+
+ @Override
+ public boolean eligible(Message message) {
+ return tracer != null && propagator != null;
+ }
+
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ // Initialize tracer from producer on first call
+ initializeIfNeeded(producer);
Review Comment:
Shouldn't we treat the first and last parts of chunked messages differently?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
+ }
+
+ @Override
+ public boolean eligible(Message message) {
+ return tracer != null && propagator != null;
+ }
+
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ // Initialize tracer from producer on first call
+ initializeIfNeeded(producer);
Review Comment:
Shouldn't we treat the first and last parts of a chunked message differently?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2668297443
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessage.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension of {@link Message} interface that supports OpenTelemetry tracing.
+ * <p>
+ * This interface allows attaching OpenTelemetry spans directly to messages,
+ * eliminating the need for external tracking via maps.
+ * <p>
+ * The span lifecycle:
+ * <ul>
+ * <li>Producer: Span is created before send and attached to the message.
+ * When the send is acknowledged, the span is retrieved and completed.</li>
+ * <li>Consumer: Span is created when message is received and attached to the message.
+ * When the message is acknowledged, the span is retrieved and completed.</li>
+ * </ul>
+ */
+public interface TraceableMessage {
+
+ /**
+ * Set the OpenTelemetry span associated with this message.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message
+ * for later retrieval when completing the span.
+ *
+ * @param span the span to associate with this message, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
So how could we add the metrics like `messaging.pulsar.acknowledgment.type` when call `onAcknowledge`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2675835327
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessageId.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension interface that allows {@link MessageId} implementations to support OpenTelemetry tracing.
+ * <p>
+ * This interface enables attaching OpenTelemetry spans directly to message IDs,
+ * allowing span retrieval in acknowledge callbacks which only receive MessageId,
+ * not the full Message object.
+ * <p>
+ * This is particularly useful for consumer-side tracing where:
+ * <ul>
+ * <li>A span is created when a message is received (in beforeConsume)</li>
+ * <li>The span is attached to the message's MessageId</li>
+ * <li>When the message is acknowledged, the span can be retrieved from the MessageId
+ * and completed, even though the acknowledge callback only provides MessageId</li>
+ * </ul>
+ */
+public interface TraceableMessageId {
+
+ /**
+ * Set the OpenTelemetry span associated with this message ID.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message ID
+ * for later retrieval in acknowledge callbacks.
+ *
+ * @param span the span to associate with this message ID, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
could you give the demo to help for the best practice?
For the message metrics, i don't have a better way to do it right now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2623635765
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
+ }
+
+ @Override
+ public boolean eligible(Message message) {
+ return tracer != null && propagator != null;
+ }
+
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ // Initialize tracer from producer on first call
+ initializeIfNeeded(producer);
Review Comment:
Shouldn't we treat the first and last parts of chunck messages differently?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2623627238
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
Review Comment:
```suggestion
// Producer will fail pending messages when it being closed, which will trigger the `onSendAcknowledgement` events
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#issuecomment-3737347471
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24873?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `46.22951%` with `164 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.46%. Comparing base ([`ff0d0eb`](https://app.codecov.io/gh/apache/pulsar/commit/ff0d0eb2ba97d96c8bc760a1dfe8255efa71d5c1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`6afcc66`](https://app.codecov.io/gh/apache/pulsar/commit/6afcc660e1db9965a397e917c1bc0d53a723451d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 14 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24873?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...impl/tracing/OpenTelemetryConsumerInterceptor.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Ftracing%2FOpenTelemetryConsumerInterceptor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3RyYWNpbmcvT3BlblRlbGVtZXRyeUNvbnN1bWVySW50ZXJjZXB0b3IuamF2YQ==) | 48.32% | [51 Missing and 26 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...impl/tracing/OpenTelemetryProducerInterceptor.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Ftracing%2FOpenTelemetryProducerInterceptor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3RyYWNpbmcvT3BlblRlbGVtZXRyeVByb2R1Y2VySW50ZXJjZXB0b3IuamF2YQ==) | 10.00% | [35 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/pulsar/client/impl/tracing/TracingContext.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Ftracing%2FTracingContext.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3RyYWNpbmcvVHJhY2luZ0NvbnRleHQuamF2YQ==) | 52.83% | [18 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ava/org/apache/pulsar/client/impl/MessageImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FMessageImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL01lc3NhZ2VJbXBsLmphdmE=) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rg/apache/pulsar/client/impl/TopicMessageImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTopicMessageImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RvcGljTWVzc2FnZUltcGwuamF2YQ==) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...apache/pulsar/client/impl/ConsumerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQnVpbGRlckltcGwuamF2YQ==) | 62.50% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...apache/pulsar/client/impl/NegativeAcksTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FNegativeAcksTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL05lZ2F0aXZlQWNrc1RyYWNrZXIuamF2YQ==) | 82.35% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...apache/pulsar/client/impl/ProducerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FProducerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1Byb2R1Y2VyQnVpbGRlckltcGwuamF2YQ==) | 62.50% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../apache/pulsar/client/impl/TopicMessageIdImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTopicMessageIdImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RvcGljTWVzc2FnZUlkSW1wbC5qYXZh) | 50.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24873 +/- ##
============================================
- Coverage 74.82% 74.46% -0.36%
- Complexity 33836 34119 +283
============================================
Files 1899 1902 +3
Lines 149656 150011 +355
Branches 17393 17461 +68
============================================
- Hits 111979 111706 -273
- Misses 28892 29395 +503
- Partials 8785 8910 +125
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24873/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24873/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.28% <2.95%> (-0.60%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24873/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.94% <2.95%> (-0.27%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24873/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.00% <46.22%> (-0.35%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24873?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...g/apache/pulsar/client/impl/ClientBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FClientBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NsaWVudEJ1aWxkZXJJbXBsLmphdmE=) | `87.16% <100.00%> (+0.13%)` | :arrow_up: |
| [...va/org/apache/pulsar/client/impl/ConsumerBase.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQmFzZS5qYXZh) | `74.84% <ø> (-0.63%)` | :arrow_down: |
| [...a/org/apache/pulsar/client/impl/MessageIdImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FMessageIdImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL01lc3NhZ2VJZEltcGwuamF2YQ==) | `90.66% <100.00%> (+0.38%)` | :arrow_up: |
| [...rg/apache/pulsar/client/impl/PulsarClientImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FPulsarClientImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1B1bHNhckNsaWVudEltcGwuamF2YQ==) | `75.47% <ø> (-0.16%)` | :arrow_down: |
| [...lsar/client/impl/conf/ClientConfigurationData.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fconf%2FClientConfigurationData.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL2NvbmYvQ2xpZW50Q29uZmlndXJhdGlvbkRhdGEuamF2YQ==) | `93.33% <ø> (ø)` | |
| [...pulsar/client/impl/metrics/InstrumentProvider.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fmetrics%2FInstrumentProvider.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL21ldHJpY3MvSW5zdHJ1bWVudFByb3ZpZGVyLmphdmE=) | `100.00% <100.00%> (ø)` | |
| [...apache/pulsar/client/impl/ConsumerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQnVpbGRlckltcGwuamF2YQ==) | `86.43% <62.50%> (-0.48%)` | :arrow_down: |
| [...apache/pulsar/client/impl/NegativeAcksTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FNegativeAcksTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL05lZ2F0aXZlQWNrc1RyYWNrZXIuamF2YQ==) | `85.85% <82.35%> (-0.89%)` | :arrow_down: |
| [...apache/pulsar/client/impl/ProducerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FProducerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1Byb2R1Y2VyQnVpbGRlckltcGwuamF2YQ==) | `82.35% <62.50%> (-0.73%)` | :arrow_down: |
| [.../apache/pulsar/client/impl/TopicMessageIdImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTopicMessageIdImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RvcGljTWVzc2FnZUlkSW1wbC5qYXZh) | `90.62% <50.00%> (-9.38%)` | :arrow_down: |
| ... and [5 more](https://app.codecov.io/gh/apache/pulsar/pull/24873?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [109 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24873/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2657825750
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/TracingContext.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Utility class for managing OpenTelemetry tracing context in Pulsar messages.
+ */
+public class TracingContext {
+
+ private static final TextMapGetter<Map<String, String>> GETTER = new TextMapGetter<Map<String, String>>() {
+ @Override
+ public Iterable<String> keys(Map<String, String> carrier) {
+ return carrier.keySet();
+ }
+
+ @Nullable
+ @Override
+ public String get(@Nullable Map<String, String> carrier, String key) {
+ return carrier != null ? carrier.get(key) : null;
+ }
+ };
+
+ private static final TextMapSetter<Map<String, String>> SETTER = (carrier, key, value) -> {
+ if (carrier != null) {
+ carrier.put(key, value);
+ }
+ };
+
+ /**
+ * Extract trace context from message properties.
+ *
+ * @param message the message to extract context from
+ * @param propagator the text map propagator to use
+ * @return the extracted context
+ */
+ public static Context extractContext(Message<?> message, TextMapPropagator propagator) {
+ if (message == null || propagator == null) {
+ return Context.current();
+ }
+ return propagator.extract(Context.current(), message.getProperties(), GETTER);
+ }
+
+ /**
+ * Inject trace context into message properties.
+ *
+ * @param messageBuilder the message builder to inject context into
+ * @param context the context to inject
+ * @param propagator the text map propagator to use
+ */
+ public static <T> void injectContext(TypedMessageBuilder<T> messageBuilder, Context context,
+ TextMapPropagator propagator) {
+ if (messageBuilder == null || context == null || propagator == null) {
+ return;
+ }
+
+ Map<String, String> carrier = new HashMap<>();
+ propagator.inject(context, carrier, SETTER);
+
+ for (Map.Entry<String, String> entry : carrier.entrySet()) {
+ messageBuilder.property(entry.getKey(), entry.getValue());
+ }
+ }
+
+ /**
+ * Create a producer span for message publishing.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param parentContext the parent context (may be null)
+ * @return the created span
+ */
+ public static Span createProducerSpan(Tracer tracer, String topic, @Nullable Context parentContext) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context context = parentContext != null ? parentContext : Context.current();
+ return tracer.spanBuilder("send " + topic)
+ .setParent(context)
+ .setSpanKind(SpanKind.PRODUCER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.operation.name", "send")
+ .startSpan();
+ }
+
+ /**
+ * Create a consumer span for message consumption.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param subscription the subscription name
+ * @param message the message being consumed
+ * @param propagator the text map propagator to use for context extraction
+ * @return the created span
+ */
+ public static Span createConsumerSpan(Tracer tracer, String topic, String subscription, Message<?> message,
+ TextMapPropagator propagator) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context parentContext = extractContext(message, propagator);
+
+ return tracer.spanBuilder("process " + topic)
+ .setParent(parentContext)
+ .setSpanKind(SpanKind.CONSUMER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.destination.subscription.name", subscription)
+ .setAttribute("messaging.operation.name", "process")
+ .setAttribute("messaging.message.id", message.getMessageId().toString())
+ .startSpan();
+ }
+
+ /**
+ * Mark a span as successful and end it.
+ *
+ * @param span the span to end
+ */
+ public static void endSpan(Span span) {
+ if (span != null && span.isRecording()) {
+ span.setStatus(StatusCode.OK);
+ span.end();
+ }
+ }
+
+ /**
+ * Mark a span as failed with an exception and end it.
+ *
+ * @param span the span to end
+ * @param throwable the exception that caused the failure
+ */
+ public static void endSpan(Span span, Throwable throwable) {
+ if (span != null && span.isRecording()) {
+ span.setStatus(StatusCode.ERROR, throwable.getMessage());
+ span.recordException(throwable);
+ span.end();
+ }
Review Comment:
Please revisit the usage of `span.isRecording()`
```suggestion
if (span != null) {
span.setStatus(StatusCode.ERROR, throwable.getMessage());
if (span.isRecording() {
span.recordException(throwable);
}
span.end();
}
```
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessageId.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension interface that allows {@link MessageId} implementations to support OpenTelemetry tracing.
+ * <p>
+ * This interface enables attaching OpenTelemetry spans directly to message IDs,
+ * allowing span retrieval in acknowledge callbacks which only receive MessageId,
+ * not the full Message object.
+ * <p>
+ * This is particularly useful for consumer-side tracing where:
+ * <ul>
+ * <li>A span is created when a message is received (in beforeConsume)</li>
+ * <li>The span is attached to the message's MessageId</li>
+ * <li>When the message is acknowledged, the span can be retrieved from the MessageId
+ * and completed, even though the acknowledge callback only provides MessageId</li>
+ * </ul>
+ */
+public interface TraceableMessageId {
+
+ /**
+ * Set the OpenTelemetry span associated with this message ID.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message ID
+ * for later retrieval in acknowledge callbacks.
+ *
+ * @param span the span to associate with this message ID, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
It doesn't seem to make sense to have this method on the public Pulsar client API. Isn't this a method that should only be used internally?
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessage.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension of {@link Message} interface that supports OpenTelemetry tracing.
+ * <p>
+ * This interface allows attaching OpenTelemetry spans directly to messages,
+ * eliminating the need for external tracking via maps.
+ * <p>
+ * The span lifecycle:
+ * <ul>
+ * <li>Producer: Span is created before send and attached to the message.
+ * When the send is acknowledged, the span is retrieved and completed.</li>
+ * <li>Consumer: Span is created when message is received and attached to the message.
+ * When the message is acknowledged, the span is retrieved and completed.</li>
+ * </ul>
+ */
+public interface TraceableMessage {
+
+ /**
+ * Set the OpenTelemetry span associated with this message.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message
+ * for later retrieval when completing the span.
+ *
+ * @param span the span to associate with this message, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
It doesn't seem to make sense to have this method on the public Pulsar client API. Isn't this a method that should only be used internally?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/TracingContext.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Utility class for managing OpenTelemetry tracing context in Pulsar messages.
+ */
+public class TracingContext {
+
+ private static final TextMapGetter<Map<String, String>> GETTER = new TextMapGetter<Map<String, String>>() {
+ @Override
+ public Iterable<String> keys(Map<String, String> carrier) {
+ return carrier.keySet();
+ }
+
+ @Nullable
+ @Override
+ public String get(@Nullable Map<String, String> carrier, String key) {
+ return carrier != null ? carrier.get(key) : null;
+ }
+ };
+
+ private static final TextMapSetter<Map<String, String>> SETTER = (carrier, key, value) -> {
+ if (carrier != null) {
+ carrier.put(key, value);
+ }
+ };
+
+ /**
+ * Extract trace context from message properties.
+ *
+ * @param message the message to extract context from
+ * @param propagator the text map propagator to use
+ * @return the extracted context
+ */
+ public static Context extractContext(Message<?> message, TextMapPropagator propagator) {
+ if (message == null || propagator == null) {
+ return Context.current();
+ }
+ return propagator.extract(Context.current(), message.getProperties(), GETTER);
+ }
+
+ /**
+ * Inject trace context into message properties.
+ *
+ * @param messageBuilder the message builder to inject context into
+ * @param context the context to inject
+ * @param propagator the text map propagator to use
+ */
+ public static <T> void injectContext(TypedMessageBuilder<T> messageBuilder, Context context,
+ TextMapPropagator propagator) {
+ if (messageBuilder == null || context == null || propagator == null) {
+ return;
+ }
+
+ Map<String, String> carrier = new HashMap<>();
+ propagator.inject(context, carrier, SETTER);
+
+ for (Map.Entry<String, String> entry : carrier.entrySet()) {
+ messageBuilder.property(entry.getKey(), entry.getValue());
+ }
+ }
+
+ /**
+ * Create a producer span for message publishing.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param parentContext the parent context (may be null)
+ * @return the created span
+ */
+ public static Span createProducerSpan(Tracer tracer, String topic, @Nullable Context parentContext) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context context = parentContext != null ? parentContext : Context.current();
+ return tracer.spanBuilder("send " + topic)
+ .setParent(context)
+ .setSpanKind(SpanKind.PRODUCER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.operation.name", "send")
+ .startSpan();
+ }
+
+ /**
+ * Create a consumer span for message consumption.
+ *
+ * @param tracer the tracer to use
+ * @param topic the topic name
+ * @param subscription the subscription name
+ * @param message the message being consumed
+ * @param propagator the text map propagator to use for context extraction
+ * @return the created span
+ */
+ public static Span createConsumerSpan(Tracer tracer, String topic, String subscription, Message<?> message,
+ TextMapPropagator propagator) {
+ if (tracer == null) {
+ return Span.getInvalid();
+ }
+
+ Context parentContext = extractContext(message, propagator);
+
+ return tracer.spanBuilder("process " + topic)
+ .setParent(parentContext)
+ .setSpanKind(SpanKind.CONSUMER)
+ .setAttribute("messaging.system", "pulsar")
+ .setAttribute("messaging.destination.name", topic)
+ .setAttribute("messaging.destination.subscription.name", subscription)
+ .setAttribute("messaging.operation.name", "process")
+ .setAttribute("messaging.message.id", message.getMessageId().toString())
+ .startSpan();
+ }
+
+ /**
+ * Mark a span as successful and end it.
+ *
+ * @param span the span to end
+ */
+ public static void endSpan(Span span) {
+ if (span != null && span.isRecording()) {
Review Comment:
the usage of `span.isRecording()` doesn't make sense here. It should only be used to skip adding some costly values to the span.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2621642566
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java:
##########
@@ -615,6 +615,44 @@ ClientBuilder authentication(String authPluginClassName, Map<String, String> aut
*/
ClientBuilder openTelemetry(io.opentelemetry.api.OpenTelemetry openTelemetry);
+ /**
+ * Enable OpenTelemetry distributed tracing.
+ *
+ * <p>When enabled, interceptors are automatically added to all producers and consumers
+ * to create spans for message publishing and consumption, and automatically propagate trace context
+ * via message properties.
+ *
+ * <p>This method is useful when OpenTelemetry is configured globally (e.g., via Java Agent or
+ * {@link io.opentelemetry.api.GlobalOpenTelemetry}) and you just want to enable tracing interceptors
+ * without explicitly setting an OpenTelemetry instance.
+ *
+ * <p>Example with Java Agent:
+ * <pre>{@code
+ * // When using -javaagent:opentelemetry-javaagent.jar
+ * PulsarClient client = PulsarClient.builder()
+ * .serviceUrl("pulsar://localhost:6650")
+ * .enableTracing(true) // Use GlobalOpenTelemetry
+ * .build();
+ * }</pre>
+ *
+ * <p>Example with GlobalOpenTelemetry:
+ * <pre>{@code
+ * // Configure GlobalOpenTelemetry elsewhere in your application
+ * GlobalOpenTelemetry.set(myOpenTelemetry);
+ *
+ * // Just enable tracing in the client
+ * PulsarClient client = PulsarClient.builder()
+ * .serviceUrl("pulsar://localhost:6650")
+ * .enableTracing(true)
+ * .build();
+ * }</pre>
+ *
+ * @param tracingEnabled whether to enable tracing (default: false)
+ * @return the client builder instance
+ * @since 4.2.0
+ */
+ ClientBuilder enableTracing(boolean tracingEnabled);
Review Comment:
收否区分类型？
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java:
##########
@@ -615,6 +615,44 @@ ClientBuilder authentication(String authPluginClassName, Map<String, String> aut
*/
ClientBuilder openTelemetry(io.opentelemetry.api.OpenTelemetry openTelemetry);
+ /**
+ * Enable OpenTelemetry distributed tracing.
+ *
+ * <p>When enabled, interceptors are automatically added to all producers and consumers
+ * to create spans for message publishing and consumption, and automatically propagate trace context
+ * via message properties.
+ *
+ * <p>This method is useful when OpenTelemetry is configured globally (e.g., via Java Agent or
+ * {@link io.opentelemetry.api.GlobalOpenTelemetry}) and you just want to enable tracing interceptors
+ * without explicitly setting an OpenTelemetry instance.
+ *
+ * <p>Example with Java Agent:
+ * <pre>{@code
+ * // When using -javaagent:opentelemetry-javaagent.jar
+ * PulsarClient client = PulsarClient.builder()
+ * .serviceUrl("pulsar://localhost:6650")
+ * .enableTracing(true) // Use GlobalOpenTelemetry
+ * .build();
+ * }</pre>
+ *
+ * <p>Example with GlobalOpenTelemetry:
+ * <pre>{@code
+ * // Configure GlobalOpenTelemetry elsewhere in your application
+ * GlobalOpenTelemetry.set(myOpenTelemetry);
+ *
+ * // Just enable tracing in the client
+ * PulsarClient client = PulsarClient.builder()
+ * .serviceUrl("pulsar://localhost:6650")
+ * .enableTracing(true)
+ * .build();
+ * }</pre>
+ *
+ * @param tracingEnabled whether to enable tracing (default: false)
+ * @return the client builder instance
+ * @since 4.2.0
+ */
+ ClientBuilder enableTracing(boolean tracingEnabled);
Review Comment:
收否区分类型？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#issuecomment-3542920743
@asafm Do you have a chance to review this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2668308811
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessageId.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension interface that allows {@link MessageId} implementations to support OpenTelemetry tracing.
+ * <p>
+ * This interface enables attaching OpenTelemetry spans directly to message IDs,
+ * allowing span retrieval in acknowledge callbacks which only receive MessageId,
+ * not the full Message object.
+ * <p>
+ * This is particularly useful for consumer-side tracing where:
+ * <ul>
+ * <li>A span is created when a message is received (in beforeConsume)</li>
+ * <li>The span is attached to the message's MessageId</li>
+ * <li>When the message is acknowledged, the span can be retrieved from the MessageId
+ * and completed, even though the acknowledge callback only provides MessageId</li>
+ * </ul>
+ */
+public interface TraceableMessageId {
+
+ /**
+ * Set the OpenTelemetry span associated with this message ID.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message ID
+ * for later retrieval in acknowledge callbacks.
+ *
+ * @param span the span to associate with this message ID, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
It's for the implementation like `onSendAcknowledgement` in OpenTelemetryProducerInterceptor to add some metrics.
Do you have any better idea?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2657864636
##########
pulsar-client/TRACING.md:
##########
@@ -0,0 +1,405 @@
+# OpenTelemetry Tracing for Pulsar Java Client
+
+This document describes how to use OpenTelemetry distributed tracing with the Pulsar Java client.
+
+## Overview
+
+The Pulsar Java client provides built-in support for OpenTelemetry distributed tracing. This allows you to:
+
+- Trace message publishing from producer to broker
+- Trace message consumption from broker to consumer
+- Propagate trace context across services via message properties
+- Extract trace context from external sources (e.g., HTTP requests)
+- Create end-to-end traces across your distributed system
+
+## Features
+
+### Producer Tracing
+
+Producer tracing creates spans for:
+- **send** - Span starts when `send()` or `sendAsync()` is called and completes when broker acknowledges receipt
+
+### Consumer Tracing
+
+Consumer tracing creates spans for:
+- **process** - Span starts when message is received and completes when message is acknowledged, negatively acknowledged, or ack timeout occurs
+
+### Trace Context Propagation
+
+Trace context is automatically propagated using W3C TraceContext format:
+- `traceparent` - Contains trace ID, span ID, and trace flags
+- `tracestate` - Contains vendor-specific trace information
+
+Context is injected into and extracted from message properties, enabling seamless trace propagation across services.
+
+## Quick Start
+
+### 1. Add Dependencies
+
+The Pulsar client already includes OpenTelemetry API dependencies. You'll need to add the SDK and exporters:
+
+```xml
+<dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk</artifactId>
+ <version>${opentelemetry.version}</version>
+</dependency>
+<dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-exporter-otlp</artifactId>
+ <version>${opentelemetry.version}</version>
+</dependency>
+```
+
+### 2. Enable Tracing
+
+There are three ways to enable tracing:
+
+#### Option 1: Using OpenTelemetry Java Agent (Easiest)
+
+```bash
+# Start your application with the Java Agent
+java -javaagent:opentelemetry-javaagent.jar \
+ -Dotel.service.name=my-service \
+ -Dotel.exporter.otlp.endpoint=http://localhost:4317 \
+ -jar your-application.jar
+```
+
+```java
+// Just enable tracing - uses GlobalOpenTelemetry from the agent
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .enableTracing(true) // That's it!
+ .build();
+```
+
+#### Option 2: With Explicit OpenTelemetry Instance
+
+```java
+OpenTelemetry openTelemetry = // configure your OpenTelemetry instance
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .openTelemetry(openTelemetry, true) // Set OpenTelemetry AND enable tracing
+ .build();
+```
+
+#### Option 3: Using GlobalOpenTelemetry
+
+```java
+// Configure GlobalOpenTelemetry once in your application
+GlobalOpenTelemetry.set(myOpenTelemetry);
+
+// Enable tracing in the client - uses GlobalOpenTelemetry
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .enableTracing(true)
+ .build();
+```
+
+**What happens when tracing is enabled:**
+- **Create spans** for producer send operations
+- **Inject trace context** into message properties automatically
+- **Create spans** for consumer receive/ack operations
+- **Extract trace context** from message properties automatically
+- Link all spans to create end-to-end distributed traces
+
+### 3. Manual Interceptor Configuration (Advanced)
+
+If you prefer manual control, you can add interceptors explicitly:
+
+```java
+import org.apache.pulsar.client.impl.tracing.OpenTelemetryProducerInterceptor;
+import org.apache.pulsar.client.impl.tracing.OpenTelemetryConsumerInterceptor;
+
+// Create client (tracing not enabled globally)
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .openTelemetry(openTelemetry)
+ .build();
+
+// Add interceptor manually to specific producer
+Producer<String> producer = client.newProducer(Schema.STRING)
+ .topic("my-topic")
+ .intercept(new OpenTelemetryProducerInterceptor())
+ .create();
+
+// Add interceptor manually to specific consumer
+Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .intercept(new OpenTelemetryConsumerInterceptor<>())
+ .subscribe();
+```
+
+## Advanced Usage
+
+### End-to-End Tracing Example
+
+This example shows how to create a complete trace from an HTTP request through Pulsar to a consumer:
+
+```java
+// Service 1: HTTP API that publishes to Pulsar
+@POST
+@Path("/order")
+public Response createOrder(@Context HttpHeaders headers, Order order) {
+ // Extract trace context from incoming HTTP request
+ Context context = TracingProducerBuilder.extractFromHeaders(
+ convertHeaders(headers));
+
+ // Publish to Pulsar with trace context
+ TracingProducerBuilder tracingBuilder = new TracingProducerBuilder();
+ producer.newMessage()
+ .value(order)
+ .let(builder -> tracingBuilder.injectContext(builder, context))
+ .send();
+
+ return Response.accepted().build();
+}
+
+// Service 2: Pulsar consumer that processes orders
+Consumer<Order> consumer = client.newConsumer(Schema.JSON(Order.class))
+ .topic("orders")
+ .subscriptionName("order-processor")
+ .intercept(new OpenTelemetryConsumerInterceptor<>())
+ .subscribe();
+
+while (true) {
+ Message<Order> msg = consumer.receive();
+ // Trace context is automatically extracted
+ // Any spans created here will be part of the same trace
+ processOrder(msg.getValue());
+ consumer.acknowledge(msg);
+}
+```
+
+### Custom Span Creation
+
+You can create custom spans during message processing:
+
+```java
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+
+Tracer tracer = GlobalOpenTelemetry.get().getTracer("my-app");
+
+Message<String> msg = consumer.receive();
+
+// Create a custom span for processing
+Span span = tracer.spanBuilder("process-message")
+ .setSpanKind(SpanKind.INTERNAL)
+ .startSpan();
+
+try (Scope scope = span.makeCurrent()) {
+ // Your processing logic
+ processMessage(msg.getValue());
+ span.setStatus(StatusCode.OK);
+} catch (Exception e) {
+ span.recordException(e);
+ span.setStatus(StatusCode.ERROR);
+ throw e;
+} finally {
+ span.end();
+ consumer.acknowledge(msg);
+}
Review Comment:
I'm just wondering how could this work. Does this PR contains tests that the custom processing span becomes a child span of the span that is created by the PulsarClient tracing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #24873:
URL: https://github.com/apache/pulsar/pull/24873
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2623669210
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
+ }
+
+ @Override
+ public boolean eligible(Message message) {
+ return tracer != null && propagator != null;
+ }
+
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ // Initialize tracer from producer on first call
+ initializeIfNeeded(producer);
Review Comment:
Please ignore this comment channel, because `sendComplete` already covered the case here: https://github.com/apache/pulsar/blob/v4.1.2/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1372-L1381
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2623669210
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/tracing/OpenTelemetryProducerInterceptor.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.TraceableMessage;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry producer interceptor that creates spans for message publishing.
+ * <p>
+ * This interceptor automatically retrieves the Tracer from the client's InstrumentProvider,
+ * ensuring consistent OpenTelemetry configuration across the client.
+ * <p>
+ * Spans are attached directly to {@link TraceableMessage} instances, eliminating the need
+ * for external span tracking via maps.
+ */
+public class OpenTelemetryProducerInterceptor implements ProducerInterceptor {
+
+ private static final Logger log = LoggerFactory.getLogger(OpenTelemetryProducerInterceptor.class);
+
+ private Tracer tracer;
+ private TextMapPropagator propagator;
+ private String topic;
+ private boolean initialized = false;
+
+ public OpenTelemetryProducerInterceptor() {
+ // Tracer and propagator will be initialized in beforeSend when we have access to the producer
+ }
+
+ /**
+ * Initialize the tracer from the producer's client.
+ * This is called lazily on the first message.
+ */
+ private void initializeIfNeeded(Producer producer) {
+ if (!initialized && producer instanceof ProducerBase<?> producerBase) {
+ PulsarClientImpl client = producerBase.getClient();
+ InstrumentProvider instrumentProvider = client.instrumentProvider();
+
+ this.tracer = instrumentProvider.getTracer();
+ this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+ this.initialized = true;
+ }
+ }
+
+ @Override
+ public void close() {
+ // No cleanup needed - spans are attached to messages
+ }
+
+ @Override
+ public boolean eligible(Message message) {
+ return tracer != null && propagator != null;
+ }
+
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ // Initialize tracer from producer on first call
+ initializeIfNeeded(producer);
Review Comment:
https://github.com/apache/pulsar/blob/v4.1.2/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1372-L1381
`sendComplete` already covered the case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2675821292
##########
pulsar-client/TRACING.md:
##########
@@ -0,0 +1,405 @@
+# OpenTelemetry Tracing for Pulsar Java Client
+
+This document describes how to use OpenTelemetry distributed tracing with the Pulsar Java client.
+
+## Overview
+
+The Pulsar Java client provides built-in support for OpenTelemetry distributed tracing. This allows you to:
+
+- Trace message publishing from producer to broker
+- Trace message consumption from broker to consumer
+- Propagate trace context across services via message properties
+- Extract trace context from external sources (e.g., HTTP requests)
+- Create end-to-end traces across your distributed system
+
+## Features
+
+### Producer Tracing
+
+Producer tracing creates spans for:
+- **send** - Span starts when `send()` or `sendAsync()` is called and completes when broker acknowledges receipt
+
+### Consumer Tracing
+
+Consumer tracing creates spans for:
+- **process** - Span starts when message is received and completes when message is acknowledged, negatively acknowledged, or ack timeout occurs
+
+### Trace Context Propagation
+
+Trace context is automatically propagated using W3C TraceContext format:
+- `traceparent` - Contains trace ID, span ID, and trace flags
+- `tracestate` - Contains vendor-specific trace information
+
+Context is injected into and extracted from message properties, enabling seamless trace propagation across services.
+
+## Quick Start
+
+### 1. Add Dependencies
+
+The Pulsar client already includes OpenTelemetry API dependencies. You'll need to add the SDK and exporters:
+
+```xml
+<dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk</artifactId>
+ <version>${opentelemetry.version}</version>
+</dependency>
+<dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-exporter-otlp</artifactId>
+ <version>${opentelemetry.version}</version>
+</dependency>
+```
+
+### 2. Enable Tracing
+
+There are three ways to enable tracing:
+
+#### Option 1: Using OpenTelemetry Java Agent (Easiest)
+
+```bash
+# Start your application with the Java Agent
+java -javaagent:opentelemetry-javaagent.jar \
+ -Dotel.service.name=my-service \
+ -Dotel.exporter.otlp.endpoint=http://localhost:4317 \
+ -jar your-application.jar
+```
+
+```java
+// Just enable tracing - uses GlobalOpenTelemetry from the agent
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .enableTracing(true) // That's it!
+ .build();
+```
+
+#### Option 2: With Explicit OpenTelemetry Instance
+
+```java
+OpenTelemetry openTelemetry = // configure your OpenTelemetry instance
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .openTelemetry(openTelemetry, true) // Set OpenTelemetry AND enable tracing
+ .build();
+```
+
+#### Option 3: Using GlobalOpenTelemetry
+
+```java
+// Configure GlobalOpenTelemetry once in your application
+GlobalOpenTelemetry.set(myOpenTelemetry);
+
+// Enable tracing in the client - uses GlobalOpenTelemetry
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .enableTracing(true)
+ .build();
+```
+
+**What happens when tracing is enabled:**
+- **Create spans** for producer send operations
+- **Inject trace context** into message properties automatically
+- **Create spans** for consumer receive/ack operations
+- **Extract trace context** from message properties automatically
+- Link all spans to create end-to-end distributed traces
+
+### 3. Manual Interceptor Configuration (Advanced)
+
+If you prefer manual control, you can add interceptors explicitly:
+
+```java
+import org.apache.pulsar.client.impl.tracing.OpenTelemetryProducerInterceptor;
+import org.apache.pulsar.client.impl.tracing.OpenTelemetryConsumerInterceptor;
+
+// Create client (tracing not enabled globally)
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .openTelemetry(openTelemetry)
+ .build();
+
+// Add interceptor manually to specific producer
+Producer<String> producer = client.newProducer(Schema.STRING)
+ .topic("my-topic")
+ .intercept(new OpenTelemetryProducerInterceptor())
+ .create();
+
+// Add interceptor manually to specific consumer
+Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .intercept(new OpenTelemetryConsumerInterceptor<>())
+ .subscribe();
+```
+
+## Advanced Usage
+
+### End-to-End Tracing Example
+
+This example shows how to create a complete trace from an HTTP request through Pulsar to a consumer:
+
+```java
+// Service 1: HTTP API that publishes to Pulsar
+@POST
+@Path("/order")
+public Response createOrder(@Context HttpHeaders headers, Order order) {
+ // Extract trace context from incoming HTTP request
+ Context context = TracingProducerBuilder.extractFromHeaders(
+ convertHeaders(headers));
+
+ // Publish to Pulsar with trace context
+ TracingProducerBuilder tracingBuilder = new TracingProducerBuilder();
+ producer.newMessage()
+ .value(order)
+ .let(builder -> tracingBuilder.injectContext(builder, context))
+ .send();
+
+ return Response.accepted().build();
+}
+
+// Service 2: Pulsar consumer that processes orders
+Consumer<Order> consumer = client.newConsumer(Schema.JSON(Order.class))
+ .topic("orders")
+ .subscriptionName("order-processor")
+ .intercept(new OpenTelemetryConsumerInterceptor<>())
+ .subscribe();
+
+while (true) {
+ Message<Order> msg = consumer.receive();
+ // Trace context is automatically extracted
+ // Any spans created here will be part of the same trace
+ processOrder(msg.getValue());
+ consumer.acknowledge(msg);
+}
+```
+
+### Custom Span Creation
+
+You can create custom spans during message processing:
+
+```java
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+
+Tracer tracer = GlobalOpenTelemetry.get().getTracer("my-app");
+
+Message<String> msg = consumer.receive();
+
+// Create a custom span for processing
+Span span = tracer.spanBuilder("process-message")
+ .setSpanKind(SpanKind.INTERNAL)
+ .startSpan();
+
+try (Scope scope = span.makeCurrent()) {
+ // Your processing logic
+ processMessage(msg.getValue());
+ span.setStatus(StatusCode.OK);
+} catch (Exception e) {
+ span.recordException(e);
+ span.setStatus(StatusCode.ERROR);
+ throw e;
+} finally {
+ span.end();
+ consumer.acknowledge(msg);
+}
Review Comment:
this is coming from `io.opentelemetry.api.trace.Tracer` and `io.opentelemetry.api.trace.Span`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2680734100
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java:
##########
@@ -0,0 +1,752 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Integration tests for OpenTelemetry tracing with real broker.
+ * Note: These tests may be timing-dependent and could be flaky in CI environments.
+ * They verify end-to-end tracing functionality with actual Pulsar broker.
+ */
+@Test(groups = "broker")
+public class OpenTelemetryTracingIntegrationTest extends BrokerTestBase {
Review Comment:
add related test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#issuecomment-3546981042
> @lhotari yes. This was the only comment I saw.
Sorry @lhotari I confused with the other one. This one I didn't review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#issuecomment-3542972867
@lhotari yes. This was the only comment I saw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2657869964
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OpenTelemetryTracingIntegrationTest.java:
##########
@@ -0,0 +1,752 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Integration tests for OpenTelemetry tracing with real broker.
+ * Note: These tests may be timing-dependent and could be flaky in CI environments.
+ * They verify end-to-end tracing functionality with actual Pulsar broker.
+ */
+@Test(groups = "broker")
+public class OpenTelemetryTracingIntegrationTest extends BrokerTestBase {
Review Comment:
This test class doesn't seem to test that "Custom Span Creation" feature described in TRACING.md works as expected. I believe that it would be expected that the custom span would be the child span of the span created by the Pulsar client's consume tracing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-446: Support Native OpenTelemetry Tracing in Pulsar Java Client [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24873:
URL: https://github.com/apache/pulsar/pull/24873#discussion_r2668352086
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TraceableMessageId.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.opentelemetry.api.trace.Span;
+
+/**
+ * Extension interface that allows {@link MessageId} implementations to support OpenTelemetry tracing.
+ * <p>
+ * This interface enables attaching OpenTelemetry spans directly to message IDs,
+ * allowing span retrieval in acknowledge callbacks which only receive MessageId,
+ * not the full Message object.
+ * <p>
+ * This is particularly useful for consumer-side tracing where:
+ * <ul>
+ * <li>A span is created when a message is received (in beforeConsume)</li>
+ * <li>The span is attached to the message's MessageId</li>
+ * <li>When the message is acknowledged, the span can be retrieved from the MessageId
+ * and completed, even though the acknowledge callback only provides MessageId</li>
+ * </ul>
+ */
+public interface TraceableMessageId {
+
+ /**
+ * Set the OpenTelemetry span associated with this message ID.
+ * <p>
+ * This method is called by tracing interceptors to attach a span to the message ID
+ * for later retrieval in acknowledge callbacks.
+ *
+ * @param span the span to associate with this message ID, or null to clear
+ */
+ void setTracingSpan(Span span);
Review Comment:
An internal interface could be used, not an interface that is exposed on the public API. I'm assuming that it doesn't make much sense to allow users of the API to call `setTracingSpan` since it's something that only the PIP-446 tracing implementation should be calling internally. Is this the correct assumption?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org