Capture pulsar message id earlier (#18803)

Author: laurit

instrumentation/pulsar/pulsar-2.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarMessagingAttributesGetter.java

@@ -11,7 +11,6 @@
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
 import java.util.List;
 import javax.annotation.Nullable;
-import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.common.naming.TopicName;
 
 final class PulsarMessagingAttributesGetter
@@ -64,12 +63,7 @@ public Long getMessageEnvelopeSize(PulsarRequest request) {
   @Nullable
   @Override
   public String getMessageId(PulsarRequest request, @Nullable Void response) {
-    Message<?> message = request.getMessage();
-    if (message.getMessageId() != null) {
-      return message.getMessageId().toString();
-    }
-
-    return null;
+    return request.getMessageId();
   }
 
   @Nullable

instrumentation/pulsar/pulsar-2.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarRequest.java

@@ -11,9 +11,11 @@
 import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData;
 import javax.annotation.Nullable;
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
 
 public class PulsarRequest extends BasePulsarRequest {
   private final Message<?> message;
+  private final String messageId;
 
   public static PulsarRequest create(Message<?> message) {
     return new PulsarRequest(message, message.getTopicName(), null);
@@ -34,9 +36,22 @@ public static PulsarRequest create(Message<?> message, ProducerData producerData
   private PulsarRequest(Message<?> message, String destination, @Nullable UrlData urlData) {
     super(destination, urlData);
     this.message = message;
+    // for producer spans message id is not available when the PulsarRequest is created, so we will
+    // try to get it later when it's available
+    MessageId id = message.getMessageId();
+    this.messageId = id != null ? id.toString() : null;
   }
 
   public Message<?> getMessage() {
     return message;
   }
+
+  @Nullable
+  public String getMessageId() {
+    if (messageId != null) {
+      return messageId;
+    }
+    MessageId id = message.getMessageId();
+    return id != null ? id.toString() : null;
+  }
 }

instrumentation/pulsar/pulsar-2.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarSingletons.java

@@ -5,8 +5,6 @@
 
 package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry;
 
-import static java.util.logging.Level.FINE;
-
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.context.Context;
@@ -32,14 +30,12 @@
 import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.VirtualFieldStore;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.logging.Logger;
 import javax.annotation.Nullable;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Messages;
 
 public class PulsarSingletons {
-  private static final Logger logger = Logger.getLogger(PulsarSingletons.class.getName());
 
   private static final String INSTRUMENTATION_NAME = "io.opentelemetry.pulsar-2.8";
 
@@ -60,8 +56,6 @@ public class PulsarSingletons {
   private static final Instrumenter<PulsarRequest, Void> producerInstrumenter =
       createProducerInstrumenter();
 
-  private static final boolean debug = Boolean.getBoolean("io.opentelemetry.pulsar-2.8.debug");
-
   public static Instrumenter<PulsarRequest, Void> consumerProcessInstrumenter() {
     return consumerProcessInstrumenter;
   }
@@ -169,9 +163,6 @@ public static Context startAndEndConsumerReceive(
     if (message == null) {
       return null;
     }
-    if (debug && message.getMessageId() == null) {
-      logger.log(FINE, "Null message id: " + message, new Exception());
-    }
     String brokerUrl = VirtualFieldStore.extract(consumer);
     PulsarRequest request = PulsarRequest.create(message, brokerUrl);
     if (!consumerReceiveInstrumenter.shouldStart(parent, request)) {