Always create a JMS consumer span (#10604)

instrumentation/jms/jms-1.1/javaagent/build.gradle.kts

@@ -36,18 +36,40 @@ testing {
         implementation("org.hornetq:hornetq-jms-client:2.4.7.Final")
         implementation("org.hornetq:hornetq-jms-server:2.4.7.Final")
       }
+
+      targets {
+        all {
+          testTask.configure {
+            jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
+          }
+        }
+      }
     }
   }
 }
 
-tasks.withType<Test>().configureEach {
-  usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
-  jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
-}
-
 tasks {
+  withType<Test>().configureEach {
+    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
+  }
+
+  val testReceiveSpansDisabled by registering(Test::class) {
+    filter {
+      includeTestsMatching("Jms1SuppressReceiveSpansTest")
+    }
+    include("**/Jms1SuppressReceiveSpansTest.*")
+  }
+
+  test {
+    filter {
+      excludeTestsMatching("Jms1SuppressReceiveSpansTest")
+    }
+    jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
+  }
+
   check {
     dependsOn(testing.suites)
+    dependsOn(testReceiveSpansDisabled)
   }
 }
 

instrumentation/jms/jms-1.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/JmsMessageConsumerInstrumentation.java

@@ -7,6 +7,7 @@ package io.opentelemetry.javaagent.instrumentation.jms.v1_1;
 
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.implementsInterface;
+import static io.opentelemetry.javaagent.instrumentation.jms.JmsReceiveSpanUtil.createReceiveSpan;
 import static io.opentelemetry.javaagent.instrumentation.jms.v1_1.JmsSingletons.consumerReceiveInstrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -14,7 +15,6 @@ import static net.bytebuddy.matcher.ElementMatchers.returns;
 import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 
 import io.opentelemetry.context.Context;
-import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
 import io.opentelemetry.instrumentation.api.internal.Timer;
 import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
@@ -75,16 +75,7 @@ public class JmsMessageConsumerInstrumentation implements TypeInstrumentation {
       MessageWithDestination request =
           MessageWithDestination.create(JavaxMessageAdapter.create(message), null);
 
-      if (consumerReceiveInstrumenter().shouldStart(parentContext, request)) {
-        InstrumenterUtil.startAndEnd(
-            consumerReceiveInstrumenter(),
-            parentContext,
-            request,
-            null,
-            throwable,
-            timer.startTime(),
-            timer.now());
-      }
+      createReceiveSpan(consumerReceiveInstrumenter(), request, timer, throwable);
     }
   }
 }

instrumentation/jms/jms-1.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/JmsSingletons.java

@@ -27,7 +27,7 @@ public final class JmsSingletons {
 
     PRODUCER_INSTRUMENTER = factory.createProducerInstrumenter();
     CONSUMER_RECEIVE_INSTRUMENTER = factory.createConsumerReceiveInstrumenter();
-    CONSUMER_PROCESS_INSTRUMENTER = factory.createConsumerProcessInstrumenter();
+    CONSUMER_PROCESS_INSTRUMENTER = factory.createConsumerProcessInstrumenter(false);
   }
 
   public static Instrumenter<MessageWithDestination, Void> producerInstrumenter() {

instrumentation/jms/jms-1.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/AbstractJms1Test.java

@@ -0,0 +1,349 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.jms.v1_1;
+
+import static io.opentelemetry.api.common.AttributeKey.stringArrayKey;
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
+import io.opentelemetry.semconv.SemanticAttributes;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQTextMessage;
+import org.assertj.core.api.AbstractAssert;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+
+abstract class AbstractJms1Test {
+  static final Logger logger = LoggerFactory.getLogger(AbstractJms1Test.class);
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
+
+  static GenericContainer<?> broker;
+  static ActiveMQConnectionFactory connectionFactory;
+  static Connection connection;
+  static Session session;
+
+  @BeforeAll
+  static void setUp() throws JMSException {
+    broker =
+        new GenericContainer<>("rmohr/activemq:latest")
+            .withExposedPorts(61616, 8161)
+            .withLogConsumer(new Slf4jLogConsumer(logger));
+    broker.start();
+
+    connectionFactory =
+        new ActiveMQConnectionFactory("tcp://localhost:" + broker.getMappedPort(61616));
+    Connection connection = connectionFactory.createConnection();
+    connection.start();
+    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+  }
+
+  @AfterAll
+  static void tearDown() throws JMSException {
+    if (session != null) {
+      session.close();
+    }
+    if (connection != null) {
+      connection.close();
+    }
+    if (broker != null) {
+      broker.close();
+    }
+  }
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void testMessageListener(
+      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
+      throws Exception {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("a message");
+
+    MessageProducer producer = session.createProducer(null);
+    cleanup.deferCleanup(producer::close);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer::close);
+
+    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
+    consumer.setMessageListener(
+        message ->
+            testing.runWithSpan(
+                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
+
+    // when
+    testing.runWithSpan("producer parent", () -> producer.send(destination, sentMessage));
+
+    // then
+    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("producer parent").hasNoParent(),
+                span ->
+                    span.hasName(destinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary)),
+                span ->
+                    span.hasName(destinationName + " process")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary)),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
+  }
+
+  @ArgumentsSource(EmptyReceiveArgumentsProvider.class)
+  @ParameterizedTest
+  void shouldNotEmitTelemetryOnEmptyReceive(
+      DestinationFactory destinationFactory, MessageReceiver receiver) throws JMSException {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer::close);
+
+    // when
+    Message message = receiver.receive(consumer);
+
+    // then
+    assertThat(message).isNull();
+
+    testing.waitForTraces(0);
+  }
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void shouldCaptureMessageHeaders(
+      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
+      throws Exception {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("a message");
+    sentMessage.setStringProperty("test_message_header", "test");
+    sentMessage.setIntProperty("test_message_int_header", 1234);
+
+    MessageProducer producer = session.createProducer(destination);
+    cleanup.deferCleanup(producer::close);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer::close);
+
+    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
+    consumer.setMessageListener(
+        message ->
+            testing.runWithSpan(
+                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
+
+    // when
+    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
+
+    // then
+    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("producer parent").hasNoParent(),
+                span ->
+                    span.hasName(destinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_header"),
+                                singletonList("test")),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_int_header"),
+                                singletonList("1234"))),
+                span ->
+                    span.hasName(destinationName + " process")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_header"),
+                                singletonList("test")),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_int_header"),
+                                singletonList("1234"))),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
+  }
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void shouldFailWhenSendingReadOnlyMessage(
+      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
+      throws Exception {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    ActiveMQTextMessage sentMessage = (ActiveMQTextMessage) session.createTextMessage("a message");
+
+    MessageProducer producer = session.createProducer(destination);
+    cleanup.deferCleanup(producer::close);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer::close);
+
+    sentMessage.setReadOnlyProperties(true);
+
+    // when
+    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
+
+    TextMessage receivedMessage = (TextMessage) consumer.receive();
+
+    // then
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String messageId = receivedMessage.getJMSMessageID();
+
+    // This will result in a logged failure because we tried to
+    // write properties in MessagePropertyTextMap when readOnlyProperties = true.
+    // As a result, the consumer span will not be linked to the producer span as we are unable to
+    // propagate the trace context as a message property.
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("producer parent").hasNoParent(),
+                span ->
+                    span.hasName(destinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary))),
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span ->
+                    span.hasName(destinationName + " receive")
+                        .hasKind(CONSUMER)
+                        .hasNoParent()
+                        .hasTotalRecordedLinks(0)
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary))));
+  }
+
+  static AttributeAssertion messagingTempDestination(boolean isTemporary) {
+    return isTemporary
+        ? equalTo(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, true)
+        : satisfies(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, AbstractAssert::isNull);
+  }
+
+  static final class EmptyReceiveArgumentsProvider implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
+      DestinationFactory topic = session -> session.createTopic("someTopic");
+      DestinationFactory queue = session -> session.createQueue("someQueue");
+      MessageReceiver receive = consumer -> consumer.receive(100);
+      MessageReceiver receiveNoWait = MessageConsumer::receiveNoWait;
+
+      return Stream.of(
+          arguments(topic, receive),
+          arguments(queue, receive),
+          arguments(topic, receiveNoWait),
+          arguments(queue, receiveNoWait));
+    }
+  }
+
+  static final class DestinationsProvider implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
+      DestinationFactory topic = session -> session.createTopic("someTopic");
+      DestinationFactory queue = session -> session.createQueue("someQueue");
+      DestinationFactory tempTopic = Session::createTemporaryTopic;
+      DestinationFactory tempQueue = Session::createTemporaryQueue;
+
+      return Stream.of(
+          arguments(topic, "someTopic", false),
+          arguments(queue, "someQueue", false),
+          arguments(tempTopic, "(temporary)", true),
+          arguments(tempQueue, "(temporary)", true));
+    }
+  }
+
+  @FunctionalInterface
+  interface DestinationFactory {
+
+    Destination create(Session session) throws JMSException;
+  }
+
+  @FunctionalInterface
+  interface MessageReceiver {
+
+    Message receive(MessageConsumer consumer) throws JMSException;
+  }
+}

instrumentation/jms/jms-1.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms1InstrumentationTest.java

@@ -5,91 +5,24 @@
 
 package io.opentelemetry.javaagent.instrumentation.jms.v1_1;
 
-import static io.opentelemetry.api.common.AttributeKey.stringArrayKey;
 import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
 import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
-import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
-import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.params.provider.Arguments.arguments;
 
-import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
-import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
-import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
-import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.sdk.trace.data.LinkData;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.semconv.SemanticAttributes;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
-import javax.jms.Connection;
 import javax.jms.Destination;
 import javax.jms.JMSException;
-import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
-import javax.jms.Session;
 import javax.jms.TextMessage;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.command.ActiveMQTextMessage;
-import org.assertj.core.api.AbstractAssert;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.ArgumentsProvider;
 import org.junit.jupiter.params.provider.ArgumentsSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
 
-public class Jms1InstrumentationTest {
-
-  static final Logger logger = LoggerFactory.getLogger(Jms1InstrumentationTest.class);
-
-  @RegisterExtension
-  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
-
-  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
-
-  static GenericContainer<?> broker;
-  static ActiveMQConnectionFactory connectionFactory;
-  static Connection connection;
-  static Session session;
-
-  @BeforeAll
-  static void setUp() throws JMSException {
-    broker =
-        new GenericContainer<>("rmohr/activemq:latest")
-            .withExposedPorts(61616, 8161)
-            .withLogConsumer(new Slf4jLogConsumer(logger));
-    broker.start();
-
-    connectionFactory =
-        new ActiveMQConnectionFactory("tcp://localhost:" + broker.getMappedPort(61616));
-    Connection connection = connectionFactory.createConnection();
-    connection.start();
-    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-  }
-
-  @AfterAll
-  static void tearDown() throws JMSException {
-    if (session != null) {
-      session.close();
-    }
-    if (connection != null) {
-      connection.close();
-    }
-    if (broker != null) {
-      broker.close();
-    }
-  }
+class Jms1InstrumentationTest extends AbstractJms1Test {
 
   @ArgumentsSource(DestinationsProvider.class)
   @ParameterizedTest
@@ -150,264 +83,4 @@ public class Jms1InstrumentationTest {
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary))));
   }
-
-  @ArgumentsSource(DestinationsProvider.class)
-  @ParameterizedTest
-  void testMessageListener(
-      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
-      throws Exception {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-    TextMessage sentMessage = session.createTextMessage("a message");
-
-    MessageProducer producer = session.createProducer(null);
-    cleanup.deferCleanup(producer::close);
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer::close);
-
-    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
-    consumer.setMessageListener(
-        message ->
-            testing.runWithSpan(
-                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
-
-    // when
-    testing.runWithSpan("producer parent", () -> producer.send(destination, sentMessage));
-
-    // then
-    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
-    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
-
-    String messageId = receivedMessage.getJMSMessageID();
-
-    testing.waitAndAssertTraces(
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("producer parent").hasNoParent(),
-                span ->
-                    span.hasName(destinationName + " publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary)),
-                span ->
-                    span.hasName(destinationName + " process")
-                        .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary)),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
-  }
-
-  @ArgumentsSource(EmptyReceiveArgumentsProvider.class)
-  @ParameterizedTest
-  void shouldNotEmitTelemetryOnEmptyReceive(
-      DestinationFactory destinationFactory, MessageReceiver receiver) throws JMSException {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer::close);
-
-    // when
-    Message message = receiver.receive(consumer);
-
-    // then
-    assertThat(message).isNull();
-
-    testing.waitForTraces(0);
-  }
-
-  @ArgumentsSource(DestinationsProvider.class)
-  @ParameterizedTest
-  void shouldCaptureMessageHeaders(
-      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
-      throws Exception {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-    TextMessage sentMessage = session.createTextMessage("a message");
-    sentMessage.setStringProperty("test_message_header", "test");
-    sentMessage.setIntProperty("test_message_int_header", 1234);
-
-    MessageProducer producer = session.createProducer(destination);
-    cleanup.deferCleanup(producer::close);
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer::close);
-
-    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
-    consumer.setMessageListener(
-        message ->
-            testing.runWithSpan(
-                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
-
-    // when
-    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
-
-    // then
-    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
-    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
-
-    String messageId = receivedMessage.getJMSMessageID();
-
-    testing.waitAndAssertTraces(
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("producer parent").hasNoParent(),
-                span ->
-                    span.hasName(destinationName + " publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_header"),
-                                singletonList("test")),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234"))),
-                span ->
-                    span.hasName(destinationName + " process")
-                        .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_header"),
-                                singletonList("test")),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234"))),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
-  }
-
-  @ArgumentsSource(DestinationsProvider.class)
-  @ParameterizedTest
-  void shouldFailWhenSendingReadOnlyMessage(
-      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
-      throws Exception {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-    ActiveMQTextMessage sentMessage = (ActiveMQTextMessage) session.createTextMessage("a message");
-
-    MessageProducer producer = session.createProducer(destination);
-    cleanup.deferCleanup(producer::close);
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer::close);
-
-    sentMessage.setReadOnlyProperties(true);
-
-    // when
-    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
-
-    TextMessage receivedMessage = (TextMessage) consumer.receive();
-
-    // then
-    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
-
-    String messageId = receivedMessage.getJMSMessageID();
-
-    // This will result in a logged failure because we tried to
-    // write properties in MessagePropertyTextMap when readOnlyProperties = true.
-    // As a result, the consumer span will not be linked to the producer span as we are unable to
-    // propagate the trace context as a message property.
-    testing.waitAndAssertTraces(
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("producer parent").hasNoParent(),
-                span ->
-                    span.hasName(destinationName + " publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary))),
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span ->
-                    span.hasName(destinationName + " receive")
-                        .hasKind(CONSUMER)
-                        .hasNoParent()
-                        .hasTotalRecordedLinks(0)
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary))));
-  }
-
-  private static AttributeAssertion messagingTempDestination(boolean isTemporary) {
-    return isTemporary
-        ? equalTo(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, true)
-        : satisfies(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, AbstractAssert::isNull);
-  }
-
-  static final class EmptyReceiveArgumentsProvider implements ArgumentsProvider {
-
-    @Override
-    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
-      DestinationFactory topic = session -> session.createTopic("someTopic");
-      DestinationFactory queue = session -> session.createQueue("someQueue");
-      MessageReceiver receive = consumer -> consumer.receive(100);
-      MessageReceiver receiveNoWait = MessageConsumer::receiveNoWait;
-
-      return Stream.of(
-          arguments(topic, receive),
-          arguments(queue, receive),
-          arguments(topic, receiveNoWait),
-          arguments(queue, receiveNoWait));
-    }
-  }
-
-  static final class DestinationsProvider implements ArgumentsProvider {
-
-    @Override
-    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
-      DestinationFactory topic = session -> session.createTopic("someTopic");
-      DestinationFactory queue = session -> session.createQueue("someQueue");
-      DestinationFactory tempTopic = Session::createTemporaryTopic;
-      DestinationFactory tempQueue = Session::createTemporaryQueue;
-
-      return Stream.of(
-          arguments(topic, "someTopic", false),
-          arguments(queue, "someQueue", false),
-          arguments(tempTopic, "(temporary)", true),
-          arguments(tempQueue, "(temporary)", true));
-    }
-  }
-
-  @FunctionalInterface
-  interface DestinationFactory {
-
-    Destination create(Session session) throws JMSException;
-  }
-
-  @FunctionalInterface
-  interface MessageReceiver {
-
-    Message receive(MessageConsumer consumer) throws JMSException;
-  }
 }

instrumentation/jms/jms-1.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms1SuppressReceiveSpansTest.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.jms.v1_1;
+
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.semconv.SemanticAttributes;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.TextMessage;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+
+class Jms1SuppressReceiveSpansTest extends AbstractJms1Test {
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void testMessageConsumer(
+      DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
+      throws JMSException {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("a message");
+
+    MessageProducer producer = session.createProducer(destination);
+    cleanup.deferCleanup(producer::close);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer::close);
+
+    // when
+    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
+
+    TextMessage receivedMessage =
+        testing.runWithSpan("consumer parent", () -> (TextMessage) consumer.receive());
+
+    // then
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("producer parent").hasNoParent(),
+                span ->
+                    span.hasName(destinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary)),
+                span ->
+                    span.hasName(destinationName + " receive")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasTotalRecordedLinks(0)
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary))),
+        trace ->
+            trace.hasSpansSatisfyingExactly(span -> span.hasName("consumer parent").hasNoParent()));
+  }
+}

instrumentation/jms/jms-3.0/javaagent/build.gradle.kts

@@ -36,7 +36,23 @@ otelJava {
 tasks {
   test {
     usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
+  }
+
+  val testReceiveSpansDisabled by registering(Test::class) {
+    filter {
+      includeTestsMatching("Jms3SuppressReceiveSpansTest")
+    }
+    include("**/Jms3SuppressReceiveSpansTest.*")
+  }
 
+  test {
+    filter {
+      excludeTestsMatching("Jms3SuppressReceiveSpansTest")
+    }
     jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
   }
+
+  check {
+    dependsOn(testReceiveSpansDisabled)
+  }
 }

instrumentation/jms/jms-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/JmsMessageConsumerInstrumentation.java

@@ -7,6 +7,7 @@ package io.opentelemetry.javaagent.instrumentation.jms.v3_0;
 
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.implementsInterface;
+import static io.opentelemetry.javaagent.instrumentation.jms.JmsReceiveSpanUtil.createReceiveSpan;
 import static io.opentelemetry.javaagent.instrumentation.jms.v3_0.JmsSingletons.consumerReceiveInstrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -14,7 +15,6 @@ import static net.bytebuddy.matcher.ElementMatchers.returns;
 import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 
 import io.opentelemetry.context.Context;
-import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
 import io.opentelemetry.instrumentation.api.internal.Timer;
 import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
@@ -75,16 +75,7 @@ public class JmsMessageConsumerInstrumentation implements TypeInstrumentation {
       MessageWithDestination request =
           MessageWithDestination.create(JakartaMessageAdapter.create(message), null);
 
-      if (consumerReceiveInstrumenter().shouldStart(parentContext, request)) {
-        InstrumenterUtil.startAndEnd(
-            consumerReceiveInstrumenter(),
-            parentContext,
-            request,
-            null,
-            throwable,
-            timer.startTime(),
-            timer.now());
-      }
+      createReceiveSpan(consumerReceiveInstrumenter(), request, timer, throwable);
     }
   }
 }

instrumentation/jms/jms-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/JmsSingletons.java

@@ -27,7 +27,7 @@ public final class JmsSingletons {
 
     PRODUCER_INSTRUMENTER = factory.createProducerInstrumenter();
     CONSUMER_RECEIVE_INSTRUMENTER = factory.createConsumerReceiveInstrumenter();
-    CONSUMER_PROCESS_INSTRUMENTER = factory.createConsumerProcessInstrumenter();
+    CONSUMER_PROCESS_INSTRUMENTER = factory.createConsumerProcessInstrumenter(false);
   }
 
   public static Instrumenter<MessageWithDestination, Void> producerInstrumenter() {

instrumentation/jms/jms-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/AbstractJms3Test.java

@@ -0,0 +1,313 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.jms.v3_0;
+
+import static io.opentelemetry.api.common.AttributeKey.stringArrayKey;
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
+import io.opentelemetry.semconv.SemanticAttributes;
+import jakarta.jms.Connection;
+import jakarta.jms.Destination;
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.Session;
+import jakarta.jms.TextMessage;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
+import org.assertj.core.api.AbstractAssert;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+abstract class AbstractJms3Test {
+  static final Logger logger = LoggerFactory.getLogger(AbstractJms3Test.class);
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
+
+  static GenericContainer<?> broker;
+  static ActiveMQConnectionFactory connectionFactory;
+  static Connection connection;
+  static Session session;
+
+  @BeforeAll
+  static void setUp() throws JMSException {
+    broker =
+        new GenericContainer<>("quay.io/artemiscloud/activemq-artemis-broker:artemis.2.27.0")
+            .withEnv("AMQ_USER", "test")
+            .withEnv("AMQ_PASSWORD", "test")
+            .withEnv("JAVA_TOOL_OPTIONS", "-Dbrokerconfig.maxDiskUsage=-1")
+            .withExposedPorts(61616, 8161)
+            .waitingFor(Wait.forLogMessage(".*Server is now live.*", 1))
+            .withStartupTimeout(Duration.ofMinutes(2))
+            .withLogConsumer(new Slf4jLogConsumer(logger));
+    broker.start();
+
+    connectionFactory =
+        new ActiveMQConnectionFactory("tcp://localhost:" + broker.getMappedPort(61616));
+    connectionFactory.setUser("test");
+    connectionFactory.setPassword("test");
+
+    connection = connectionFactory.createConnection();
+    connection.start();
+
+    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+  }
+
+  @AfterAll
+  static void tearDown() throws JMSException {
+    if (session != null) {
+      session.close();
+    }
+    if (connection != null) {
+      connection.close();
+    }
+    if (connectionFactory != null) {
+      connectionFactory.close();
+    }
+    if (broker != null) {
+      broker.close();
+    }
+  }
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void testMessageListener(DestinationFactory destinationFactory, boolean isTemporary)
+      throws Exception {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("hello there");
+
+    MessageProducer producer = session.createProducer(null);
+    cleanup.deferCleanup(producer);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer);
+
+    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
+    consumer.setMessageListener(
+        message ->
+            testing.runWithSpan(
+                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
+
+    // when
+    testing.runWithSpan("parent", () -> producer.send(destination, sentMessage));
+
+    // then
+    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String actualDestinationName = ((ActiveMQDestination) destination).getName();
+    // artemis consumers don't know whether the destination is temporary or not
+    String producerDestinationName = isTemporary ? "(temporary)" : actualDestinationName;
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasNoParent(),
+                span ->
+                    span.hasName(producerDestinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                producerDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary)),
+                span ->
+                    span.hasName(actualDestinationName + " process")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                actualDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId)),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
+  }
+
+  @ArgumentsSource(EmptyReceiveArgumentsProvider.class)
+  @ParameterizedTest
+  void shouldNotEmitTelemetryOnEmptyReceive(
+      DestinationFactory destinationFactory, MessageReceiver receiver) throws JMSException {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer);
+
+    // when
+    Message message = receiver.receive(consumer);
+
+    // then
+    assertThat(message).isNull();
+
+    testing.waitForTraces(0);
+  }
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void shouldCaptureMessageHeaders(DestinationFactory destinationFactory, boolean isTemporary)
+      throws Exception {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("hello there");
+    sentMessage.setStringProperty("test_message_header", "test");
+    sentMessage.setIntProperty("test_message_int_header", 1234);
+
+    MessageProducer producer = session.createProducer(destination);
+    cleanup.deferCleanup(producer);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer);
+
+    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
+    consumer.setMessageListener(
+        message ->
+            testing.runWithSpan(
+                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
+
+    // when
+    testing.runWithSpan("parent", () -> producer.send(sentMessage));
+
+    // then
+    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String actualDestinationName = ((ActiveMQDestination) destination).getName();
+    // artemis consumers don't know whether the destination is temporary or not
+    String producerDestinationName = isTemporary ? "(temporary)" : actualDestinationName;
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasNoParent(),
+                span ->
+                    span.hasName(producerDestinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                producerDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_header"),
+                                singletonList("test")),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_int_header"),
+                                singletonList("1234"))),
+                span ->
+                    span.hasName(actualDestinationName + " process")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                actualDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_header"),
+                                singletonList("test")),
+                            equalTo(
+                                stringArrayKey("messaging.header.test_message_int_header"),
+                                singletonList("1234"))),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
+  }
+
+  static AttributeAssertion messagingTempDestination(boolean isTemporary) {
+    return isTemporary
+        ? equalTo(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, true)
+        : satisfies(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, AbstractAssert::isNull);
+  }
+
+  static final class EmptyReceiveArgumentsProvider implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
+      DestinationFactory topic = session -> session.createTopic("someTopic");
+      DestinationFactory queue = session -> session.createQueue("someQueue");
+      MessageReceiver receive = consumer -> consumer.receive(100);
+      MessageReceiver receiveNoWait = MessageConsumer::receiveNoWait;
+
+      return Stream.of(
+          arguments(topic, receive),
+          arguments(queue, receive),
+          arguments(topic, receiveNoWait),
+          arguments(queue, receiveNoWait));
+    }
+  }
+
+  static final class DestinationsProvider implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
+      DestinationFactory topic = session -> session.createTopic("someTopic");
+      DestinationFactory queue = session -> session.createQueue("someQueue");
+      DestinationFactory tempTopic = Session::createTemporaryTopic;
+      DestinationFactory tempQueue = Session::createTemporaryQueue;
+
+      return Stream.of(
+          arguments(topic, false),
+          arguments(queue, false),
+          arguments(tempTopic, true),
+          arguments(tempQueue, true));
+    }
+  }
+
+  @FunctionalInterface
+  interface DestinationFactory {
+
+    Destination create(Session session) throws JMSException;
+  }
+
+  @FunctionalInterface
+  interface MessageReceiver {
+
+    Message receive(MessageConsumer consumer) throws JMSException;
+  }
+}

instrumentation/jms/jms-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/Jms3InstrumentationTest.java

@@ -5,105 +5,25 @@
 
 package io.opentelemetry.javaagent.instrumentation.jms.v3_0;
 
-import static io.opentelemetry.api.common.AttributeKey.stringArrayKey;
 import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
 import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
-import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
-import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.params.provider.Arguments.arguments;
 
-import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
-import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
-import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
-import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.sdk.trace.data.LinkData;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.semconv.SemanticAttributes;
-import jakarta.jms.Connection;
 import jakarta.jms.Destination;
 import jakarta.jms.JMSException;
-import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
 import jakarta.jms.MessageProducer;
-import jakarta.jms.Session;
 import jakarta.jms.TextMessage;
-import java.time.Duration;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
-import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
 import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
-import org.assertj.core.api.AbstractAssert;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.ArgumentsProvider;
 import org.junit.jupiter.params.provider.ArgumentsSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.strategy.Wait;
 
-class Jms3InstrumentationTest {
-
-  static final Logger logger = LoggerFactory.getLogger(Jms3InstrumentationTest.class);
-
-  @RegisterExtension
-  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
-
-  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
-
-  static GenericContainer<?> broker;
-  static ActiveMQConnectionFactory connectionFactory;
-  static Connection connection;
-  static Session session;
-
-  @BeforeAll
-  static void setUp() throws JMSException {
-    broker =
-        new GenericContainer<>("quay.io/artemiscloud/activemq-artemis-broker:artemis.2.27.0")
-            .withEnv("AMQ_USER", "test")
-            .withEnv("AMQ_PASSWORD", "test")
-            .withEnv("JAVA_TOOL_OPTIONS", "-Dbrokerconfig.maxDiskUsage=-1")
-            .withExposedPorts(61616, 8161)
-            .waitingFor(Wait.forLogMessage(".*Server is now live.*", 1))
-            .withStartupTimeout(Duration.ofMinutes(2))
-            .withLogConsumer(new Slf4jLogConsumer(logger));
-    broker.start();
-
-    connectionFactory =
-        new ActiveMQConnectionFactory("tcp://localhost:" + broker.getMappedPort(61616));
-    connectionFactory.setUser("test");
-    connectionFactory.setPassword("test");
-
-    connection = connectionFactory.createConnection();
-    connection.start();
-
-    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-  }
-
-  @AfterAll
-  static void tearDown() throws JMSException {
-    if (session != null) {
-      session.close();
-    }
-    if (connection != null) {
-      connection.close();
-    }
-    if (connectionFactory != null) {
-      connectionFactory.close();
-    }
-    if (broker != null) {
-      broker.close();
-    }
-  }
+class Jms3InstrumentationTest extends AbstractJms3Test {
 
   @ArgumentsSource(DestinationsProvider.class)
   @ParameterizedTest
@@ -169,214 +89,4 @@ class Jms3InstrumentationTest {
                             equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId))));
   }
-
-  @ArgumentsSource(DestinationsProvider.class)
-  @ParameterizedTest
-  void testMessageListener(DestinationFactory destinationFactory, boolean isTemporary)
-      throws Exception {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-    TextMessage sentMessage = session.createTextMessage("hello there");
-
-    MessageProducer producer = session.createProducer(null);
-    cleanup.deferCleanup(producer);
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer);
-
-    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
-    consumer.setMessageListener(
-        message ->
-            testing.runWithSpan(
-                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
-
-    // when
-    testing.runWithSpan("parent", () -> producer.send(destination, sentMessage));
-
-    // then
-    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
-    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
-
-    String actualDestinationName = ((ActiveMQDestination) destination).getName();
-    // artemis consumers don't know whether the destination is temporary or not
-    String producerDestinationName = isTemporary ? "(temporary)" : actualDestinationName;
-    String messageId = receivedMessage.getJMSMessageID();
-
-    testing.waitAndAssertTraces(
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("parent").hasNoParent(),
-                span ->
-                    span.hasName(producerDestinationName + " publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(
-                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
-                                producerDestinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary)),
-                span ->
-                    span.hasName(actualDestinationName + " process")
-                        .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(
-                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
-                                actualDestinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId)),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
-  }
-
-  @ArgumentsSource(EmptyReceiveArgumentsProvider.class)
-  @ParameterizedTest
-  void shouldNotEmitTelemetryOnEmptyReceive(
-      DestinationFactory destinationFactory, MessageReceiver receiver) throws JMSException {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer);
-
-    // when
-    Message message = receiver.receive(consumer);
-
-    // then
-    assertThat(message).isNull();
-
-    testing.waitForTraces(0);
-  }
-
-  @ArgumentsSource(DestinationsProvider.class)
-  @ParameterizedTest
-  void shouldCaptureMessageHeaders(DestinationFactory destinationFactory, boolean isTemporary)
-      throws Exception {
-
-    // given
-    Destination destination = destinationFactory.create(session);
-    TextMessage sentMessage = session.createTextMessage("hello there");
-    sentMessage.setStringProperty("test_message_header", "test");
-    sentMessage.setIntProperty("test_message_int_header", 1234);
-
-    MessageProducer producer = session.createProducer(destination);
-    cleanup.deferCleanup(producer);
-    MessageConsumer consumer = session.createConsumer(destination);
-    cleanup.deferCleanup(consumer);
-
-    CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
-    consumer.setMessageListener(
-        message ->
-            testing.runWithSpan(
-                "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
-
-    // when
-    testing.runWithSpan("parent", () -> producer.send(sentMessage));
-
-    // then
-    TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
-    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
-
-    String actualDestinationName = ((ActiveMQDestination) destination).getName();
-    // artemis consumers don't know whether the destination is temporary or not
-    String producerDestinationName = isTemporary ? "(temporary)" : actualDestinationName;
-    String messageId = receivedMessage.getJMSMessageID();
-
-    testing.waitAndAssertTraces(
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("parent").hasNoParent(),
-                span ->
-                    span.hasName(producerDestinationName + " publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(
-                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
-                                producerDestinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            messagingTempDestination(isTemporary),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_header"),
-                                singletonList("test")),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234"))),
-                span ->
-                    span.hasName(actualDestinationName + " process")
-                        .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(
-                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
-                                actualDestinationName),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
-                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_header"),
-                                singletonList("test")),
-                            equalTo(
-                                stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234"))),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
-  }
-
-  private static AttributeAssertion messagingTempDestination(boolean isTemporary) {
-    return isTemporary
-        ? equalTo(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, true)
-        : satisfies(SemanticAttributes.MESSAGING_DESTINATION_TEMPORARY, AbstractAssert::isNull);
-  }
-
-  static final class EmptyReceiveArgumentsProvider implements ArgumentsProvider {
-
-    @Override
-    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
-      DestinationFactory topic = session -> session.createTopic("someTopic");
-      DestinationFactory queue = session -> session.createQueue("someQueue");
-      MessageReceiver receive = consumer -> consumer.receive(100);
-      MessageReceiver receiveNoWait = MessageConsumer::receiveNoWait;
-
-      return Stream.of(
-          arguments(topic, receive),
-          arguments(queue, receive),
-          arguments(topic, receiveNoWait),
-          arguments(queue, receiveNoWait));
-    }
-  }
-
-  static final class DestinationsProvider implements ArgumentsProvider {
-
-    @Override
-    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
-      DestinationFactory topic = session -> session.createTopic("someTopic");
-      DestinationFactory queue = session -> session.createQueue("someQueue");
-      DestinationFactory tempTopic = Session::createTemporaryTopic;
-      DestinationFactory tempQueue = Session::createTemporaryQueue;
-
-      return Stream.of(
-          arguments(topic, false),
-          arguments(queue, false),
-          arguments(tempTopic, true),
-          arguments(tempQueue, true));
-    }
-  }
-
-  @FunctionalInterface
-  interface DestinationFactory {
-
-    Destination create(Session session) throws JMSException;
-  }
-
-  @FunctionalInterface
-  interface MessageReceiver {
-
-    Message receive(MessageConsumer consumer) throws JMSException;
-  }
 }

instrumentation/jms/jms-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/Jms3SuppressReceiveSpansTest.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.jms.v3_0;
+
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.semconv.SemanticAttributes;
+import jakarta.jms.Destination;
+import jakarta.jms.JMSException;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.TextMessage;
+import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+
+class Jms3SuppressReceiveSpansTest extends AbstractJms3Test {
+
+  @ArgumentsSource(DestinationsProvider.class)
+  @ParameterizedTest
+  void testMessageConsumer(DestinationFactory destinationFactory, boolean isTemporary)
+      throws JMSException {
+
+    // given
+    Destination destination = destinationFactory.create(session);
+    TextMessage sentMessage = session.createTextMessage("hello there");
+
+    MessageProducer producer = session.createProducer(destination);
+    cleanup.deferCleanup(producer);
+    MessageConsumer consumer = session.createConsumer(destination);
+    cleanup.deferCleanup(consumer);
+
+    // when
+    testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
+
+    TextMessage receivedMessage =
+        testing.runWithSpan("consumer parent", () -> (TextMessage) consumer.receive());
+
+    // then
+    assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
+
+    String actualDestinationName = ((ActiveMQDestination) destination).getName();
+    // artemis consumers don't know whether the destination is temporary or not
+    String producerDestinationName = isTemporary ? "(temporary)" : actualDestinationName;
+    String messageId = receivedMessage.getJMSMessageID();
+
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("producer parent").hasNoParent(),
+                span ->
+                    span.hasName(producerDestinationName + " publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                producerDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
+                            messagingTempDestination(isTemporary)),
+                span ->
+                    span.hasName(actualDestinationName + " receive")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasTotalRecordedLinks(0)
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                actualDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
+                            equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId))),
+        trace ->
+            trace.hasSpansSatisfyingExactly(span -> span.hasName("consumer parent").hasNoParent()));
+  }
+}

instrumentation/jms/jms-common/bootstrap/build.gradle.kts

@@ -0,0 +1,3 @@
+plugins {
+  id("otel.javaagent-bootstrap")
+}

instrumentation/jms/jms-common/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/jms/JmsReceiveContextHolder.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.bootstrap.jms;
+
+import static io.opentelemetry.context.ContextKey.named;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
+import io.opentelemetry.context.ImplicitContextKeyed;
+import javax.annotation.Nullable;
+
+public final class JmsReceiveContextHolder implements ImplicitContextKeyed {
+  private static final ContextKey<JmsReceiveContextHolder> KEY =
+      named("opentelemetry-jms-receive-context");
+
+  private Context receiveContext;
+
+  private JmsReceiveContextHolder() {}
+
+  public static Context init(Context context) {
+    if (context.get(KEY) != null) {
+      return context;
+    }
+    return context.with(new JmsReceiveContextHolder());
+  }
+
+  public static void set(Context receiveContext) {
+    JmsReceiveContextHolder holder = receiveContext.get(KEY);
+    if (holder != null) {
+      holder.receiveContext = receiveContext;
+    }
+  }
+
+  @Nullable
+  public static Context getReceiveContext(Context context) {
+    JmsReceiveContextHolder holder = context.get(KEY);
+    return holder != null ? holder.receiveContext : null;
+  }
+
+  @Override
+  public Context storeInContext(Context context) {
+    return context.with(KEY, this);
+  }
+}

instrumentation/jms/jms-common/javaagent/build.gradle.kts

@@ -5,4 +5,6 @@ plugins {
 dependencies {
   compileOnly("com.google.auto.value:auto-value-annotations")
   annotationProcessor("com.google.auto.value:auto-value")
+
+  bootstrap(project(":instrumentation:jms:jms-common:bootstrap"))
 }

instrumentation/jms/jms-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/JmsInstrumenterFactory.java

@@ -14,6 +14,7 @@ import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.Messagin
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
 import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor;
 import java.util.List;
@@ -59,30 +60,41 @@ public final class JmsInstrumenterFactory {
     JmsMessageAttributesGetter getter = JmsMessageAttributesGetter.INSTANCE;
     MessageOperation operation = MessageOperation.RECEIVE;
 
-    // MessageConsumer does not do context propagation
-    return Instrumenter.<MessageWithDestination, Void>builder(
-            openTelemetry,
-            instrumentationName,
-            MessagingSpanNameExtractor.create(getter, operation))
-        .addAttributesExtractor(createMessagingAttributesExtractor(operation))
-        .setEnabled(messagingReceiveInstrumentationEnabled)
-        .addSpanLinksExtractor(
-            new PropagatorBasedSpanLinksExtractor<>(
-                openTelemetry.getPropagators().getTextMapPropagator(),
-                MessagePropertyGetter.INSTANCE))
-        .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
+    InstrumenterBuilder<MessageWithDestination, Void> builder =
+        Instrumenter.<MessageWithDestination, Void>builder(
+                openTelemetry,
+                instrumentationName,
+                MessagingSpanNameExtractor.create(getter, operation))
+            .addAttributesExtractor(createMessagingAttributesExtractor(operation));
+    if (messagingReceiveInstrumentationEnabled) {
+      builder.addSpanLinksExtractor(
+          new PropagatorBasedSpanLinksExtractor<>(
+              openTelemetry.getPropagators().getTextMapPropagator(),
+              MessagePropertyGetter.INSTANCE));
+    }
+    return builder.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
   }
 
-  public Instrumenter<MessageWithDestination, Void> createConsumerProcessInstrumenter() {
+  public Instrumenter<MessageWithDestination, Void> createConsumerProcessInstrumenter(
+      boolean canHaveReceiveInstrumentation) {
     JmsMessageAttributesGetter getter = JmsMessageAttributesGetter.INSTANCE;
     MessageOperation operation = MessageOperation.PROCESS;
 
-    return Instrumenter.<MessageWithDestination, Void>builder(
-            openTelemetry,
-            instrumentationName,
-            MessagingSpanNameExtractor.create(getter, operation))
-        .addAttributesExtractor(createMessagingAttributesExtractor(operation))
-        .buildConsumerInstrumenter(MessagePropertyGetter.INSTANCE);
+    InstrumenterBuilder<MessageWithDestination, Void> builder =
+        Instrumenter.<MessageWithDestination, Void>builder(
+                openTelemetry,
+                instrumentationName,
+                MessagingSpanNameExtractor.create(getter, operation))
+            .addAttributesExtractor(createMessagingAttributesExtractor(operation));
+    if (canHaveReceiveInstrumentation && messagingReceiveInstrumentationEnabled) {
+      builder.addSpanLinksExtractor(
+          new PropagatorBasedSpanLinksExtractor<>(
+              openTelemetry.getPropagators().getTextMapPropagator(),
+              MessagePropertyGetter.INSTANCE));
+      return builder.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
+    } else {
+      return builder.buildConsumerInstrumenter(MessagePropertyGetter.INSTANCE);
+    }
   }
 
   private AttributesExtractor<MessageWithDestination, Void> createMessagingAttributesExtractor(

instrumentation/jms/jms-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/JmsReceiveSpanUtil.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.jms;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
+import io.opentelemetry.instrumentation.api.internal.Timer;
+import io.opentelemetry.javaagent.bootstrap.internal.ExperimentalConfig;
+import io.opentelemetry.javaagent.bootstrap.jms.JmsReceiveContextHolder;
+
+public final class JmsReceiveSpanUtil {
+  private static final ContextPropagators propagators = GlobalOpenTelemetry.getPropagators();
+  private static final boolean receiveInstrumentationEnabled =
+      ExperimentalConfig.get().messagingReceiveInstrumentationEnabled();
+
+  public static void createReceiveSpan(
+      Instrumenter<MessageWithDestination, Void> receiveInstrumenter,
+      MessageWithDestination request,
+      Timer timer,
+      Throwable throwable) {
+    Context parentContext = Context.current();
+    // if receive instrumentation is not enabled we'll use the producer as parent
+    if (!receiveInstrumentationEnabled) {
+      parentContext =
+          propagators
+              .getTextMapPropagator()
+              .extract(parentContext, request, MessagePropertyGetter.INSTANCE);
+    }
+
+    if (receiveInstrumenter.shouldStart(parentContext, request)) {
+      Context receiveContext =
+          InstrumenterUtil.startAndEnd(
+              receiveInstrumenter,
+              parentContext,
+              request,
+              null,
+              throwable,
+              timer.startTime(),
+              timer.now());
+      JmsReceiveContextHolder.set(receiveContext);
+    }
+  }
+
+  private JmsReceiveSpanUtil() {}
+}

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/build.gradle.kts

@@ -14,6 +14,7 @@ muzzle {
 }
 
 dependencies {
+  bootstrap(project(":instrumentation:jms:jms-common:bootstrap"))
   implementation(project(":instrumentation:jms:jms-common:javaagent"))
   implementation(project(":instrumentation:jms:jms-1.1:javaagent"))
   library("org.springframework:spring-jms:2.0")

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v2_0/AbstractPollingMessageListenerContainerInstrumentation.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0;
+
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0.SpringJmsSingletons.isReceiveTelemetryEnabled;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.bootstrap.jms.JmsReceiveContextHolder;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+public class AbstractPollingMessageListenerContainerInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.springframework.jms.listener.AbstractPollingMessageListenerContainer");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("receiveAndExecute"), this.getClass().getName() + "$ReceiveAndExecuteAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ReceiveAndExecuteAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static Scope onEnter() {
+      if (isReceiveTelemetryEnabled()) {
+        Context context = JmsReceiveContextHolder.init(Java8BytecodeBridge.currentContext());
+        return context.makeCurrent();
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(@Advice.Enter Scope scope) {
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+    }
+  }
+}

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v2_0/JmsDestinationAccessorInstrumentation.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0;
+
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0.SpringJmsSingletons.isReceiveTelemetryEnabled;
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0.SpringJmsSingletons.receiveInstrumenter;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.returns;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
+import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+public class JmsDestinationAccessorInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.springframework.jms.support.destination.JmsDestinationAccessor");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("receiveFromConsumer").and(returns(named("javax.jms.Message"))),
+        this.getClass().getName() + "$ReceiveAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ReceiveAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static Scope onEnter() {
+      if (isReceiveTelemetryEnabled()) {
+        return null;
+      }
+      // suppress receive span creation in jms instrumentation
+      Context context =
+          InstrumenterUtil.suppressSpan(
+              receiveInstrumenter(), Java8BytecodeBridge.currentContext(), null);
+      return context.makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(@Advice.Enter Scope scope) {
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+    }
+  }
+}

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v2_0/SpringJmsInstrumentationModule.java

@@ -6,11 +6,11 @@
 package io.opentelemetry.javaagent.instrumentation.spring.jms.v2_0;
 
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
-import static java.util.Collections.singletonList;
 
 import com.google.auto.service.AutoService;
 import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import java.util.Arrays;
 import java.util.List;
 import net.bytebuddy.matcher.ElementMatcher;
 
@@ -29,6 +29,9 @@ public class SpringJmsInstrumentationModule extends InstrumentationModule {
 
   @Override
   public List<TypeInstrumentation> typeInstrumentations() {
-    return singletonList(new SpringJmsMessageListenerInstrumentation());
+    return Arrays.asList(
+        new SpringJmsMessageListenerInstrumentation(),
+        new JmsDestinationAccessorInstrumentation(),
+        new AbstractPollingMessageListenerContainerInstrumentation());
   }
 }

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v2_0/SpringJmsMessageListenerInstrumentation.java

@@ -16,6 +16,7 @@ import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.bootstrap.jms.JmsReceiveContextHolder;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import io.opentelemetry.javaagent.instrumentation.jms.MessageWithDestination;
@@ -59,6 +60,10 @@ public class SpringJmsMessageListenerInstrumentation implements TypeInstrumentat
         @Advice.Local("otelScope") Scope scope) {
 
       Context parentContext = Java8BytecodeBridge.currentContext();
+      Context receiveContext = JmsReceiveContextHolder.getReceiveContext(parentContext);
+      if (receiveContext != null) {
+        parentContext = receiveContext;
+      }
       request = MessageWithDestination.create(JavaxMessageAdapter.create(message), null);
 
       if (!listenerInstrumenter().shouldStart(parentContext, request)) {

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v2_0/SpringJmsSingletons.java

@@ -14,14 +14,32 @@ import io.opentelemetry.javaagent.instrumentation.jms.MessageWithDestination;
 public final class SpringJmsSingletons {
   private static final String INSTRUMENTATION_NAME = "io.opentelemetry.spring-jms-2.0";
 
-  private static final Instrumenter<MessageWithDestination, Void> LISTENER_INSTRUMENTER =
-      new JmsInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
-          .setCapturedHeaders(ExperimentalConfig.get().getMessagingHeaders())
-          .createConsumerProcessInstrumenter();
+  private static final boolean RECEIVE_TELEMETRY_ENABLED =
+      ExperimentalConfig.get().messagingReceiveInstrumentationEnabled();
+  private static final Instrumenter<MessageWithDestination, Void> LISTENER_INSTRUMENTER;
+  private static final Instrumenter<MessageWithDestination, Void> RECEIVE_INSTRUMENTER;
+
+  static {
+    JmsInstrumenterFactory factory =
+        new JmsInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
+            .setCapturedHeaders(ExperimentalConfig.get().getMessagingHeaders())
+            .setMessagingReceiveInstrumentationEnabled(RECEIVE_TELEMETRY_ENABLED);
+
+    LISTENER_INSTRUMENTER = factory.createConsumerProcessInstrumenter(true);
+    RECEIVE_INSTRUMENTER = factory.createConsumerReceiveInstrumenter();
+  }
+
+  public static boolean isReceiveTelemetryEnabled() {
+    return RECEIVE_TELEMETRY_ENABLED;
+  }
 
   public static Instrumenter<MessageWithDestination, Void> listenerInstrumenter() {
     return LISTENER_INSTRUMENTER;
   }
 
+  public static Instrumenter<MessageWithDestination, Void> receiveInstrumenter() {
+    return RECEIVE_INSTRUMENTER;
+  }
+
   private SpringJmsSingletons() {}
 }

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/test/groovy/SpringListenerTest.groovy

@@ -26,16 +26,18 @@ class SpringListenerTest extends AgentInstrumentationSpecification {
 
     template.convertAndSend("SpringListenerJms2", "a message")
 
+    def producer
     expect:
     assertTraces(2) {
-      traces.sort(orderByRootSpanKind(CONSUMER, PRODUCER))
+      traces.sort(orderByRootSpanKind(PRODUCER, CONSUMER))
 
       trace(0, 1) {
-        consumerSpan(it, 0, "SpringListenerJms2", "", null, "receive")
+        producerSpan(it, 0, "SpringListenerJms2")
+        producer = span(0)
       }
       trace(1, 2) {
-        producerSpan(it, 0, "SpringListenerJms2")
-        consumerSpan(it, 1, "SpringListenerJms2", "", span(0), "process")
+        consumerSpan(it, 0, "SpringListenerJms2", "", null, producer, "receive")
+        consumerSpan(it, 1, "SpringListenerJms2", "", span(0), producer, "process")
       }
     }
 
@@ -70,15 +72,20 @@ class SpringListenerTest extends AgentInstrumentationSpecification {
   // passing messageId = null will verify message.id is not captured,
   // passing messageId = "" will verify message.id is captured (but won't verify anything about the value),
   // any other value for messageId will verify that message.id is captured and has that same value
-  static consumerSpan(TraceAssert trace, int index, String destinationName, String messageId, Object parentOrLinkedSpan, String operation, boolean testHeaders = false) {
+  static consumerSpan(TraceAssert trace, int index, String destinationName, String messageId, Object parent, Object linkedSpan, String operation, boolean testHeaders = false) {
     trace.span(index) {
       name destinationName + " " + operation
       kind CONSUMER
-      if (parentOrLinkedSpan != null) {
-        childOf((SpanData) parentOrLinkedSpan)
+      if (parent != null) {
+        childOf((SpanData) parent)
       } else {
         hasNoParent()
       }
+      if (linkedSpan != null) {
+        hasLink((SpanData) linkedSpan)
+      } else {
+        hasNoLinks()
+      }
       attributes {
         "$SemanticAttributes.MESSAGING_SYSTEM" "jms"
         "$SemanticAttributes.MESSAGING_DESTINATION_NAME" destinationName

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/test/groovy/SpringTemplateTest.groovy

@@ -94,11 +94,13 @@ class SpringTemplateTest extends AgentInstrumentationSpecification {
     expect:
     receivedMessage.text == messageText
     assertTraces(2) {
+      def producer
       trace(0, 1) {
         producerSpan(it, 0, destinationName)
+        producer = span(0)
       }
       trace(1, 1) {
-        consumerSpan(it, 0, destinationName, receivedMessage.getJMSMessageID(), null, "receive")
+        consumerSpan(it, 0, destinationName, receivedMessage.getJMSMessageID(), null, producer, "receive")
       }
     }
 
@@ -128,22 +130,26 @@ class SpringTemplateTest extends AgentInstrumentationSpecification {
     receivedMessage.text == "responded!"
     assertTraces(4) {
       traces.sort(orderByRootSpanName(
-        "$destinationName receive",
         "$destinationName publish",
-        "(temporary) receive",
-        "(temporary) publish"))
+        "$destinationName receive",
+        "(temporary) publish",
+        "(temporary) receive"))
 
+      def producer
       trace(0, 1) {
-        consumerSpan(it, 0, destinationName, msgId.get(), null, "receive")
+        producerSpan(it, 0, destinationName)
+        producer = span(0)
       }
       trace(1, 1) {
-        producerSpan(it, 0, destinationName)
+        consumerSpan(it, 0, destinationName, msgId.get(), null, producer, "receive")
       }
+      def tempProducer
       trace(2, 1) {
-        consumerSpan(it, 0, "(temporary)", receivedMessage.getJMSMessageID(), null, "receive")
+        producerSpan(it, 0, "(temporary)")
+        tempProducer = span(0)
       }
       trace(3, 1) {
-        producerSpan(it, 0, "(temporary)")
+        consumerSpan(it, 0, "(temporary)", receivedMessage.getJMSMessageID(), null, tempProducer, "receive")
       }
     }
 
@@ -167,11 +173,13 @@ class SpringTemplateTest extends AgentInstrumentationSpecification {
     expect:
     receivedMessage.text == messageText
     assertTraces(2) {
+      def producer
       trace(0, 1) {
         producerSpan(it, 0, destinationName, true)
+        producer = span(0)
       }
       trace(1, 1) {
-        consumerSpan(it, 0, destinationName, receivedMessage.getJMSMessageID(), null, "receive", true)
+        consumerSpan(it, 0, destinationName, receivedMessage.getJMSMessageID(), null, producer,"receive", true)
       }
     }
 

instrumentation/spring/spring-jms/spring-jms-2.0/javaagent/src/testReceiveSpansDisabled/groovy/SpringListenerSuppressReceiveSpansTest.groovy

@@ -23,7 +23,7 @@ class SpringListenerSuppressReceiveSpansTest extends AgentInstrumentationSpecifi
     assertTraces(1) {
       trace(0, 2) {
         SpringListenerTest.producerSpan(it, 0, "SpringListenerJms2")
-        SpringListenerTest.consumerSpan(it, 1, "SpringListenerJms2", "", span(0), "process")
+        SpringListenerTest.consumerSpan(it, 1, "SpringListenerJms2", "", span(0), null, "process")
       }
     }
 

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/build.gradle.kts

@@ -14,6 +14,7 @@ muzzle {
 }
 
 dependencies {
+  bootstrap(project(":instrumentation:jms:jms-common:bootstrap"))
   implementation(project(":instrumentation:jms:jms-common:javaagent"))
   implementation(project(":instrumentation:jms:jms-3.0:javaagent"))
 
@@ -34,9 +35,25 @@ otelJava {
 }
 
 tasks {
-  test {
+  withType<Test>().configureEach {
     usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
+  }
+
+  val testReceiveSpansDisabled by registering(Test::class) {
+    filter {
+      includeTestsMatching("SpringListenerSuppressReceiveSpansTest")
+    }
+    include("**/SpringListenerSuppressReceiveSpansTest.*")
+  }
 
+  test {
+    filter {
+      excludeTestsMatching("SpringListenerSuppressReceiveSpansTest")
+    }
     jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
   }
+
+  check {
+    dependsOn(testReceiveSpansDisabled)
+  }
 }

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/AbstractPollingMessageListenerContainerInstrumentation.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0;
+
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0.SpringJmsSingletons.isReceiveTelemetryEnabled;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.bootstrap.jms.JmsReceiveContextHolder;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+public class AbstractPollingMessageListenerContainerInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.springframework.jms.listener.AbstractPollingMessageListenerContainer");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("receiveAndExecute"), this.getClass().getName() + "$ReceiveAndExecuteAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ReceiveAndExecuteAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static Scope onEnter() {
+      if (isReceiveTelemetryEnabled()) {
+        Context context = JmsReceiveContextHolder.init(Java8BytecodeBridge.currentContext());
+        return context.makeCurrent();
+      }
+      return null;
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(@Advice.Enter Scope scope) {
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+    }
+  }
+}

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/JmsDestinationAccessorInstrumentation.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0;
+
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0.SpringJmsSingletons.isReceiveTelemetryEnabled;
+import static io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0.SpringJmsSingletons.receiveInstrumenter;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.returns;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
+import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+public class JmsDestinationAccessorInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return named("org.springframework.jms.support.destination.JmsDestinationAccessor");
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("receiveFromConsumer").and(returns(named("jakarta.jms.Message"))),
+        this.getClass().getName() + "$ReceiveAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ReceiveAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static Scope onEnter() {
+      if (isReceiveTelemetryEnabled()) {
+        return null;
+      }
+      // suppress receive span creation in jms instrumentation
+      Context context =
+          InstrumenterUtil.suppressSpan(
+              receiveInstrumenter(), Java8BytecodeBridge.currentContext(), null);
+      return context.makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(@Advice.Enter Scope scope) {
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+    }
+  }
+}

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/SpringJmsInstrumentationModule.java

@@ -6,7 +6,7 @@
 package io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0;
 
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
-import static java.util.Collections.singletonList;
+import static java.util.Arrays.asList;
 import static net.bytebuddy.matcher.ElementMatchers.not;
 
 import com.google.auto.service.AutoService;
@@ -30,6 +30,9 @@ public class SpringJmsInstrumentationModule extends InstrumentationModule {
 
   @Override
   public List<TypeInstrumentation> typeInstrumentations() {
-    return singletonList(new SpringJmsMessageListenerInstrumentation());
+    return asList(
+        new SpringJmsMessageListenerInstrumentation(),
+        new JmsDestinationAccessorInstrumentation(),
+        new AbstractPollingMessageListenerContainerInstrumentation());
   }
 }

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/SpringJmsMessageListenerInstrumentation.java

@@ -16,6 +16,7 @@ import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.bootstrap.jms.JmsReceiveContextHolder;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import io.opentelemetry.javaagent.instrumentation.jms.MessageWithDestination;
@@ -59,6 +60,10 @@ public class SpringJmsMessageListenerInstrumentation implements TypeInstrumentat
         @Advice.Local("otelScope") Scope scope) {
 
       Context parentContext = Java8BytecodeBridge.currentContext();
+      Context receiveContext = JmsReceiveContextHolder.getReceiveContext(parentContext);
+      if (receiveContext != null) {
+        parentContext = receiveContext;
+      }
       request = MessageWithDestination.create(JakartaMessageAdapter.create(message), null);
 
       if (!listenerInstrumenter().shouldStart(parentContext, request)) {

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/SpringJmsSingletons.java

@@ -14,14 +14,32 @@ import io.opentelemetry.javaagent.instrumentation.jms.MessageWithDestination;
 public final class SpringJmsSingletons {
   private static final String INSTRUMENTATION_NAME = "io.opentelemetry.spring-jms-6.0";
 
-  private static final Instrumenter<MessageWithDestination, Void> LISTENER_INSTRUMENTER =
-      new JmsInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
-          .setCapturedHeaders(ExperimentalConfig.get().getMessagingHeaders())
-          .createConsumerProcessInstrumenter();
+  private static final boolean RECEIVE_TELEMETRY_ENABLED =
+      ExperimentalConfig.get().messagingReceiveInstrumentationEnabled();
+  private static final Instrumenter<MessageWithDestination, Void> LISTENER_INSTRUMENTER;
+  private static final Instrumenter<MessageWithDestination, Void> RECEIVE_INSTRUMENTER;
+
+  static {
+    JmsInstrumenterFactory factory =
+        new JmsInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
+            .setCapturedHeaders(ExperimentalConfig.get().getMessagingHeaders())
+            .setMessagingReceiveInstrumentationEnabled(RECEIVE_TELEMETRY_ENABLED);
+
+    LISTENER_INSTRUMENTER = factory.createConsumerProcessInstrumenter(true);
+    RECEIVE_INSTRUMENTER = factory.createConsumerReceiveInstrumenter();
+  }
+
+  public static boolean isReceiveTelemetryEnabled() {
+    return RECEIVE_TELEMETRY_ENABLED;
+  }
 
   public static Instrumenter<MessageWithDestination, Void> listenerInstrumenter() {
     return LISTENER_INSTRUMENTER;
   }
 
+  public static Instrumenter<MessageWithDestination, Void> receiveInstrumenter() {
+    return RECEIVE_INSTRUMENTER;
+  }
+
   private SpringJmsSingletons() {}
 }

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/AbstractSpringJmsListenerTest.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import jakarta.jms.ConnectionFactory;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.jms.core.JmsTemplate;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+abstract class AbstractSpringJmsListenerTest {
+  static final Logger logger = LoggerFactory.getLogger(AbstractSpringJmsListenerTest.class);
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
+
+  static GenericContainer<?> broker;
+
+  @BeforeAll
+  static void setUp() {
+    broker =
+        new GenericContainer<>("quay.io/artemiscloud/activemq-artemis-broker:artemis.2.27.0")
+            .withEnv("AMQ_USER", "test")
+            .withEnv("AMQ_PASSWORD", "test")
+            .withEnv("JAVA_TOOL_OPTIONS", "-Dbrokerconfig.maxDiskUsage=-1")
+            .withExposedPorts(61616, 8161)
+            .waitingFor(Wait.forLogMessage(".*Server is now live.*", 1))
+            .withStartupTimeout(Duration.ofMinutes(2))
+            .withLogConsumer(new Slf4jLogConsumer(logger));
+    broker.start();
+  }
+
+  @AfterAll
+  static void tearDown() {
+    if (broker != null) {
+      broker.close();
+    }
+  }
+
+  @ArgumentsSource(SpringJmsListenerTest.ConfigClasses.class)
+  @ParameterizedTest
+  @SuppressWarnings("unchecked")
+  void testSpringJmsListener(Class<?> configClass)
+      throws ExecutionException, InterruptedException, TimeoutException {
+    // given
+    SpringApplication app = new SpringApplication(configClass);
+    app.setDefaultProperties(defaultConfig());
+    ConfigurableApplicationContext applicationContext = app.run();
+    cleanup.deferCleanup(applicationContext);
+
+    JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean(ConnectionFactory.class));
+    String message = "hello there";
+
+    // when
+    testing.runWithSpan("parent", () -> jmsTemplate.convertAndSend("spring-jms-listener", message));
+
+    // then
+    CompletableFuture<String> receivedMessage =
+        applicationContext.getBean("receivedMessage", CompletableFuture.class);
+    assertThat(receivedMessage.get(10, TimeUnit.SECONDS)).isEqualTo(message);
+
+    assertSpringJmsListener();
+  }
+
+  abstract void assertSpringJmsListener();
+
+  static Map<String, Object> defaultConfig() {
+    Map<String, Object> props = new HashMap<>();
+    props.put("spring.jmx.enabled", false);
+    props.put("spring.main.web-application-type", "none");
+    props.put("test.broker-url", "tcp://localhost:" + broker.getMappedPort(61616));
+    return props;
+  }
+
+  static final class ConfigClasses implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
+      return Stream.of(
+          arguments(AnnotatedListenerConfig.class), arguments(ManualListenerConfig.class));
+    }
+  }
+}

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/SpringJmsListenerTest.java

@@ -14,115 +14,69 @@ import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equal
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.params.provider.Arguments.arguments;
 
-import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
-import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
-import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.semconv.SemanticAttributes;
 import jakarta.jms.ConnectionFactory;
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.stream.Stream;
+import java.util.concurrent.atomic.AtomicReference;
 import org.assertj.core.api.AbstractStringAssert;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.ArgumentsProvider;
 import org.junit.jupiter.params.provider.ArgumentsSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.jms.core.JmsTemplate;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.strategy.Wait;
 
-class SpringJmsListenerTest {
-
-  static final Logger logger = LoggerFactory.getLogger(SpringJmsListenerTest.class);
-
-  @RegisterExtension
-  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
-
-  @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
-
-  static GenericContainer<?> broker;
-
-  @BeforeAll
-  static void setUp() {
-    broker =
-        new GenericContainer<>("quay.io/artemiscloud/activemq-artemis-broker:artemis.2.27.0")
-            .withEnv("AMQ_USER", "test")
-            .withEnv("AMQ_PASSWORD", "test")
-            .withEnv("JAVA_TOOL_OPTIONS", "-Dbrokerconfig.maxDiskUsage=-1")
-            .withExposedPorts(61616, 8161)
-            .waitingFor(Wait.forLogMessage(".*Server is now live.*", 1))
-            .withStartupTimeout(Duration.ofMinutes(2))
-            .withLogConsumer(new Slf4jLogConsumer(logger));
-    broker.start();
-  }
-
-  @AfterAll
-  static void tearDown() {
-    if (broker != null) {
-      broker.close();
-    }
-  }
-
-  @ArgumentsSource(ConfigClasses.class)
-  @ParameterizedTest
-  @SuppressWarnings("unchecked")
-  void testSpringJmsListener(Class<?> configClass)
-      throws ExecutionException, InterruptedException, TimeoutException {
-    // given
-    SpringApplication app = new SpringApplication(configClass);
-    app.setDefaultProperties(defaultConfig());
-    ConfigurableApplicationContext applicationContext = app.run();
-    cleanup.deferCleanup(applicationContext);
-
-    JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean(ConnectionFactory.class));
-    String message = "hello there";
-
-    // when
-    testing.runWithSpan("parent", () -> jmsTemplate.convertAndSend("spring-jms-listener", message));
-
-    // then
-    CompletableFuture<String> receivedMessage =
-        applicationContext.getBean("receivedMessage", CompletableFuture.class);
-    assertThat(receivedMessage.get(10, TimeUnit.SECONDS)).isEqualTo(message);
+class SpringJmsListenerTest extends AbstractSpringJmsListenerTest {
 
+  @Override
+  void assertSpringJmsListener() {
+    AtomicReference<SpanData> producerSpan = new AtomicReference<>();
     testing.waitAndAssertSortedTraces(
         orderByRootSpanKind(INTERNAL, CONSUMER),
+        trace -> {
+          trace.hasSpansSatisfyingExactly(
+              span -> span.hasName("parent").hasNoParent(),
+              span ->
+                  span.hasName("spring-jms-listener publish")
+                      .hasKind(PRODUCER)
+                      .hasParent(trace.getSpan(0))
+                      .hasAttributesSatisfyingExactly(
+                          equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                          equalTo(
+                              SemanticAttributes.MESSAGING_DESTINATION_NAME, "spring-jms-listener"),
+                          equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                          satisfies(
+                              SemanticAttributes.MESSAGING_MESSAGE_ID,
+                              AbstractStringAssert::isNotBlank)));
+
+          producerSpan.set(trace.getSpan(1));
+        },
         trace ->
             trace.hasSpansSatisfyingExactly(
-                span -> span.hasName("parent").hasNoParent(),
                 span ->
-                    span.hasName("spring-jms-listener publish")
-                        .hasKind(PRODUCER)
-                        .hasParent(trace.getSpan(0))
+                    span.hasName("spring-jms-listener receive")
+                        .hasKind(CONSUMER)
+                        .hasNoParent()
+                        .hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(
                                 SemanticAttributes.MESSAGING_DESTINATION_NAME,
                                 "spring-jms-listener"),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
                             satisfies(
                                 SemanticAttributes.MESSAGING_MESSAGE_ID,
                                 AbstractStringAssert::isNotBlank)),
                 span ->
                     span.hasName("spring-jms-listener process")
                         .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
+                        .hasParent(trace.getSpan(0))
+                        .hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(
@@ -132,22 +86,7 @@ class SpringJmsListenerTest {
                             satisfies(
                                 SemanticAttributes.MESSAGING_MESSAGE_ID,
                                 AbstractStringAssert::isNotBlank)),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))),
-        trace ->
-            trace.hasSpansSatisfyingExactly(
-                span ->
-                    span.hasName("spring-jms-listener receive")
-                        .hasKind(CONSUMER)
-                        .hasNoParent()
-                        .hasAttributesSatisfyingExactly(
-                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
-                            equalTo(
-                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
-                                "spring-jms-listener"),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
-                            satisfies(
-                                SemanticAttributes.MESSAGING_MESSAGE_ID,
-                                AbstractStringAssert::isNotBlank))));
+                span -> span.hasName("consumer").hasParent(trace.getSpan(1))));
   }
 
   @ArgumentsSource(ConfigClasses.class)
@@ -205,17 +144,19 @@ class SpringJmsListenerTest {
                                 singletonList("test")),
                             equalTo(
                                 stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234"))),
+                                singletonList("1234")))),
+        trace ->
+            trace.hasSpansSatisfyingExactly(
                 span ->
-                    span.hasName("spring-jms-listener process")
+                    span.hasName("spring-jms-listener receive")
                         .hasKind(CONSUMER)
-                        .hasParent(trace.getSpan(1))
+                        .hasNoParent()
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(
                                 SemanticAttributes.MESSAGING_DESTINATION_NAME,
                                 "spring-jms-listener"),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
                             satisfies(
                                 SemanticAttributes.MESSAGING_MESSAGE_ID,
                                 AbstractStringAssert::isNotBlank),
@@ -225,19 +166,16 @@ class SpringJmsListenerTest {
                             equalTo(
                                 stringArrayKey("messaging.header.test_message_int_header"),
                                 singletonList("1234"))),
-                span -> span.hasName("consumer").hasParent(trace.getSpan(2))),
-        trace ->
-            trace.hasSpansSatisfyingExactly(
                 span ->
-                    span.hasName("spring-jms-listener receive")
+                    span.hasName("spring-jms-listener process")
                         .hasKind(CONSUMER)
-                        .hasNoParent()
+                        .hasParent(trace.getSpan(0))
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(
                                 SemanticAttributes.MESSAGING_DESTINATION_NAME,
                                 "spring-jms-listener"),
-                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
                             satisfies(
                                 SemanticAttributes.MESSAGING_MESSAGE_ID,
                                 AbstractStringAssert::isNotBlank),
@@ -246,23 +184,7 @@ class SpringJmsListenerTest {
                                 singletonList("test")),
                             equalTo(
                                 stringArrayKey("messaging.header.test_message_int_header"),
-                                singletonList("1234")))));
-  }
-
-  private static Map<String, Object> defaultConfig() {
-    Map<String, Object> props = new HashMap<>();
-    props.put("spring.jmx.enabled", false);
-    props.put("spring.main.web-application-type", "none");
-    props.put("test.broker-url", "tcp://localhost:" + broker.getMappedPort(61616));
-    return props;
-  }
-
-  static final class ConfigClasses implements ArgumentsProvider {
-
-    @Override
-    public Stream<? extends Arguments> provideArguments(ExtensionContext context) {
-      return Stream.of(
-          arguments(AnnotatedListenerConfig.class), arguments(ManualListenerConfig.class));
-    }
+                                singletonList("1234"))),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(1))));
   }
 }

instrumentation/spring/spring-jms/spring-jms-6.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/jms/v6_0/SpringListenerSuppressReceiveSpansTest.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.spring.jms.v6_0;
+
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
+
+import io.opentelemetry.semconv.SemanticAttributes;
+import org.assertj.core.api.AbstractStringAssert;
+
+class SpringListenerSuppressReceiveSpansTest extends AbstractSpringJmsListenerTest {
+
+  @Override
+  void assertSpringJmsListener() {
+    testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(
+                span -> span.hasName("parent").hasNoParent(),
+                span ->
+                    span.hasName("spring-jms-listener publish")
+                        .hasKind(PRODUCER)
+                        .hasParent(trace.getSpan(0))
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                "spring-jms-listener"),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
+                            satisfies(
+                                SemanticAttributes.MESSAGING_MESSAGE_ID,
+                                AbstractStringAssert::isNotBlank)),
+                span ->
+                    span.hasName("spring-jms-listener process")
+                        .hasKind(CONSUMER)
+                        .hasParent(trace.getSpan(1))
+                        .hasTotalRecordedLinks(0)
+                        .hasAttributesSatisfyingExactly(
+                            equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
+                            equalTo(
+                                SemanticAttributes.MESSAGING_DESTINATION_NAME,
+                                "spring-jms-listener"),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "process"),
+                            satisfies(
+                                SemanticAttributes.MESSAGING_MESSAGE_ID,
+                                AbstractStringAssert::isNotBlank)),
+                span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
+  }
+}

javaagent-tooling/src/main/java/io/opentelemetry/javaagent/tooling/ignore/AdditionalLibraryIgnoredTypesConfigurer.java

@@ -174,7 +174,8 @@ public class AdditionalLibraryIgnoredTypesConfigurer implements IgnoredTypesConf
         .ignoreClass("org.springframework.jms.")
         .allowClass("org.springframework.jms.listener.")
         .allowClass(
-            "org.springframework.jms.config.JmsListenerEndpointRegistry$AggregatingCallback");
+            "org.springframework.jms.config.JmsListenerEndpointRegistry$AggregatingCallback")
+        .allowClass("org.springframework.jms.support.destination.JmsDestinationAccessor");
 
     builder
         .ignoreClass("org.springframework.util.")

settings.gradle.kts

@@ -324,6 +324,7 @@ include(":instrumentation:jetty-httpclient:jetty-httpclient-9.2:library")
 include(":instrumentation:jetty-httpclient:jetty-httpclient-9.2:testing")
 include(":instrumentation:jms:jms-1.1:javaagent")
 include(":instrumentation:jms:jms-3.0:javaagent")
+include(":instrumentation:jms:jms-common:bootstrap")
 include(":instrumentation:jms:jms-common:javaagent")
 include(":instrumentation:jms:jms-common:javaagent-unit-tests")
 include(":instrumentation:jmx-metrics:javaagent")