Repository: pinpoint-apm/pinpoint
Issue: Add support for spring-amqp-rabbit 2.1.0.RELEASE
Labels: enhancement
module:plugin

#### Description
Constructor for `org.springframework.amqp.rabbit.support.Delivery` got an extra parameter in 2.1.0.RELEASE.
`org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$ConsumerDecorator` has been removed in 2.1.0.RELEASE.

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/rabbitmq/src/main/java/com/navercorp/pinpoint/plugin/rabbitmq/client/RabbitMQClientPlugin.java b/plugins/rabbitmq/src/main/java/com/navercorp/pinpoint/plugin/rabbitmq/client/RabbitMQClientPlugin.java
index fe0a042d21ef..930eb5699c53 100644
--- a/plugins/rabbitmq/src/main/java/com/navercorp/pinpoint/plugin/rabbitmq/client/RabbitMQClientPlugin.java
+++ b/plugins/rabbitmq/src/main/java/com/navercorp/pinpoint/plugin/rabbitmq/client/RabbitMQClientPlugin.java
@@ -384,6 +384,11 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader loader, Strin
                     if (constructor != null) {
                         constructor.addScopedInterceptor("com.navercorp.pinpoint.plugin.rabbitmq.client.interceptor.DeliveryConstructInterceptor", RabbitMQClientConstants.RABBITMQ_CONSUMER_SCOPE, ExecutionPolicy.INTERNAL);
                     }
+                    // spring-rabbit 2.1.0+
+                    InstrumentMethod constructor_2_1_0 = target.getConstructor("java.lang.String", "com.rabbitmq.client.Envelope", "com.rabbitmq.client.AMQP$BasicProperties", "byte[]", "java.lang.String");
+                    if (constructor_2_1_0 != null) {
+                        constructor_2_1_0.addScopedInterceptor("com.navercorp.pinpoint.plugin.rabbitmq.client.interceptor.DeliveryConstructInterceptor", RabbitMQClientConstants.RABBITMQ_CONSUMER_SCOPE, ExecutionPolicy.INTERNAL);
+                    }
                     target.addField(AsyncContextAccessor.class.getName());
                     return target.toBytecode();
                 }


Test patch:
diff --git a/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java
index a7e725ce595b..e369b45d0f4c 100644
--- a/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java
+++ b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java
@@ -49,7 +49,7 @@
  * <tt>BlockingQueueConsumer$ConsumerDecorator</tt> added in 2.0.3 has been backported to 1.7.7 as well.
  *
  * @author HyunGil Jeong
- * @see SpringAmqpRabbit_2_0_3_to_2_x_IT
+ * @see SpringAmqpRabbit_2_0_3_to_2_1_0_IT
  */
 @RunWith(PinpointPluginTestSuite.class)
 @PinpointAgent(AgentPath.PATH)
diff --git a/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_1_0_IT.java
similarity index 99%
rename from agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java
rename to agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_1_0_IT.java
index 8f5d91d85c85..1c3e5d174ea1 100644
--- a/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java
+++ b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_1_0_IT.java
@@ -59,9 +59,9 @@
 @RunWith(PinpointPluginTestSuite.class)
 @PinpointAgent(AgentPath.PATH)
 @PinpointConfig("rabbitmq/client/pinpoint-rabbitmq.config")
-@Dependency({"org.springframework.amqp:spring-rabbit:[2.0.3.RELEASE,)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
+@Dependency({"org.springframework.amqp:spring-rabbit:[2.0.3.RELEASE,2.1.0.RELEASE)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
 @JvmVersion(8)
-public class SpringAmqpRabbit_2_0_3_to_2_x_IT {
+public class SpringAmqpRabbit_2_0_3_to_2_1_0_IT {
 
     private static final TestBroker BROKER = new TestBroker();
     private static final TestApplicationContext CONTEXT = new TestApplicationContext();
diff --git a/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_1_x_to_2_x_IT.java b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_1_x_to_2_x_IT.java
new file mode 100644
index 000000000000..43f292662fd4
--- /dev/null
+++ b/agent-it/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_1_x_to_2_x_IT.java
@@ -0,0 +1,310 @@
+/*
+ * Copyright 2018 NAVER Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.navercorp.pinpoint.plugin.jdk7.rabbitmq.spring;
+
+import com.navercorp.pinpoint.bootstrap.plugin.test.Expectations;
+import com.navercorp.pinpoint.bootstrap.plugin.test.ExpectedTrace;
+import com.navercorp.pinpoint.bootstrap.plugin.test.PluginTestVerifier;
+import com.navercorp.pinpoint.common.trace.ServiceType;
+import com.navercorp.pinpoint.plugin.AgentPath;
+import com.navercorp.pinpoint.plugin.jdk7.rabbitmq.util.RabbitMQTestConstants;
+import com.navercorp.pinpoint.plugin.jdk7.rabbitmq.util.TestBroker;
+import com.navercorp.pinpoint.test.plugin.Dependency;
+import com.navercorp.pinpoint.test.plugin.JvmVersion;
+import com.navercorp.pinpoint.test.plugin.PinpointAgent;
+import com.navercorp.pinpoint.test.plugin.PinpointConfig;
+import com.navercorp.pinpoint.test.plugin.PinpointPluginTestSuite;
+import com.navercorp.test.pinpoint.plugin.rabbitmq.PropagationMarker;
+import com.navercorp.test.pinpoint.plugin.rabbitmq.spring.config.CommonConfig;
+import com.navercorp.test.pinpoint.plugin.rabbitmq.spring.config.MessageListenerConfig_Post_1_4_0;
+import com.navercorp.test.pinpoint.plugin.rabbitmq.spring.config.ReceiverConfig_Post_1_6_0;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.impl.AMQCommand;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.amqp.core.Message;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+
+/**
+ * <p>Spring-amqp rabbit 2.1.0 removed previously added <tt>BlockingQueueConsumer$ConsumerDecorator</tt>.
+ *
+ * @author HyunGil Jeong
+ */
+@RunWith(PinpointPluginTestSuite.class)
+@PinpointAgent(AgentPath.PATH)
+@PinpointConfig("rabbitmq/client/pinpoint-rabbitmq.config")
+@Dependency({"org.springframework.amqp:spring-rabbit:[2.1.0.RELEASE,)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
+@JvmVersion(8)
+public class SpringAmqpRabbit_2_1_x_to_2_x_IT {
+
+    private static final TestBroker BROKER = new TestBroker();
+    private static final TestApplicationContext CONTEXT = new TestApplicationContext();
+
+    private final SpringAmqpRabbitTestRunner testRunner = new SpringAmqpRabbitTestRunner(CONTEXT);
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        BROKER.start();
+        CONTEXT.init(
+                CommonConfig.class,
+                MessageListenerConfig_Post_1_4_0.class,
+                ReceiverConfig_Post_1_6_0.class);
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() {
+        CONTEXT.close();
+        BROKER.shutdown();
+    }
+
+    @Test
+    public void testPush() throws Exception {
+        final String remoteAddress = testRunner.getRemoteAddress();
+
+        Class<?> rabbitTemplateClass = Class.forName("org.springframework.amqp.rabbit.core.RabbitTemplate");
+        Method rabbitTemplateConvertAndSend = rabbitTemplateClass.getDeclaredMethod("convertAndSend", String.class, String.class, Object.class);
+        ExpectedTrace rabbitTemplateConvertAndSendTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateConvertAndSend); // method
+        // automatic recovery deliberately disabled as Spring has it's own recovery mechanism
+        Class<?> channelNClass = Class.forName("com.rabbitmq.client.impl.ChannelN");
+        Method channelNBasicPublish = channelNClass.getDeclaredMethod("basicPublish", String.class, String.class, boolean.class, boolean.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace channelNBasicPublishTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                channelNBasicPublish, // method
+                null, // rpc
+                remoteAddress, // endPoint
+                "exchange-" + RabbitMQTestConstants.EXCHANGE, // destinationId
+                Expectations.annotation("rabbitmq.exchange", RabbitMQTestConstants.EXCHANGE),
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PUSH));
+        ExpectedTrace rabbitMqConsumerInvocationTrace = Expectations.root(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                "RabbitMQ Consumer Invocation", // method
+                "rabbitmq://exchange=" + RabbitMQTestConstants.EXCHANGE, // rpc
+                null, // endPoint (collected but API to retrieve local address is not available in all versions, so skip)
+                remoteAddress, // remoteAddress
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PUSH));
+        Class<?> consumerDispatcherClass = Class.forName("com.rabbitmq.client.impl.ConsumerDispatcher");
+        Method consumerDispatcherHandleDelivery = consumerDispatcherClass.getDeclaredMethod("handleDelivery", Consumer.class, String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace consumerDispatcherHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                consumerDispatcherHandleDelivery); // method
+        ExpectedTrace asynchronousInvocationTrace = Expectations.event(
+                ServiceType.ASYNC.getName(),
+                "Asynchronous Invocation");
+        Class<?> blockingQueueConsumerInternalConsumerClass = Class.forName("org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$InternalConsumer");
+        Method blockingQueueConsumerInternalConsumerHandleDelivery = blockingQueueConsumerInternalConsumerClass.getDeclaredMethod("handleDelivery", String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace blockingQueueConsumerInternalConsumerHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                blockingQueueConsumerInternalConsumerHandleDelivery);
+        Class<?> deliveryClass = Class.forName("org.springframework.amqp.rabbit.support.Delivery");
+        Constructor<?> deliveryConstructor = deliveryClass.getDeclaredConstructor(String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class, String.class);
+        ExpectedTrace deliveryConstructorTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                deliveryConstructor);
+        Class<?> abstractMessageListenerContainerClass = Class.forName("org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer");
+        Method abstractMessageListenerContainerExecuteListener = abstractMessageListenerContainerClass.getDeclaredMethod("executeListener", Channel.class, Message.class);
+        ExpectedTrace abstractMessageListenerContainerExecuteListenerTrace = Expectations.event(
+                ServiceType.INTERNAL_METHOD.getName(),
+                abstractMessageListenerContainerExecuteListener);
+        Class<?> propagationMarkerClass = PropagationMarker.class;
+        Method propagationMarkerMark = propagationMarkerClass.getDeclaredMethod("mark");
+        ExpectedTrace markTrace = Expectations.event(
+                ServiceType.INTERNAL_METHOD.getName(),
+                propagationMarkerMark);
+
+        ExpectedTrace[] expectedTraces = {
+                rabbitTemplateConvertAndSendTrace,
+                channelNBasicPublishTrace,
+                rabbitMqConsumerInvocationTrace,
+                consumerDispatcherHandleDeliveryTrace,
+                asynchronousInvocationTrace,
+                blockingQueueConsumerInternalConsumerHandleDeliveryTrace,
+                deliveryConstructorTrace,
+                asynchronousInvocationTrace,
+                abstractMessageListenerContainerExecuteListenerTrace,
+                markTrace
+        };
+
+        final PluginTestVerifier verifier = testRunner.runPush(expectedTraces.length);
+
+        verifier.verifyTrace(expectedTraces);
+        verifier.verifyTraceCount(0);
+    }
+
+    @Test
+    public void testPull() throws Exception {
+        final String remoteAddress = testRunner.getRemoteAddress();
+
+        Class<?> rabbitTemplateClass = Class.forName("org.springframework.amqp.rabbit.core.RabbitTemplate");
+        // verify queue-initiated traces
+        Method rabbitTemplateConvertAndSend = rabbitTemplateClass.getDeclaredMethod("convertAndSend", String.class, String.class, Object.class);
+        ExpectedTrace rabbitTemplateConvertAndSendTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateConvertAndSend); // method
+        // automatic recovery deliberately disabled as Spring has it's own recovery mechanism
+        Class<?> channelNClass = Class.forName("com.rabbitmq.client.impl.ChannelN");
+        Method channelNBasicPublish = channelNClass.getDeclaredMethod("basicPublish", String.class, String.class, boolean.class, boolean.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace channelNBasicPublishTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                channelNBasicPublish, // method
+                null, // rpc
+                remoteAddress, // endPoint
+                "exchange-" + RabbitMQTestConstants.EXCHANGE, // destinationId
+                Expectations.annotation("rabbitmq.exchange", RabbitMQTestConstants.EXCHANGE),
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PULL));
+        ExpectedTrace rabbitMqConsumerInvocationTrace = Expectations.root(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                "RabbitMQ Consumer Invocation", // method
+                "rabbitmq://exchange=" + RabbitMQTestConstants.EXCHANGE, // rpc
+                null, // endPoint (collected but API to retrieve local address is not available in all versions, so skip)
+                remoteAddress, // remoteAddress
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PULL));
+        Class<?> amqChannelClass = Class.forName("com.rabbitmq.client.impl.AMQChannel");
+        Method amqChannelHandleCompleteInboundCommand = amqChannelClass.getDeclaredMethod("handleCompleteInboundCommand", AMQCommand.class);
+        ExpectedTrace amqChannelHandleCompleteInboundCommandTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // method
+                amqChannelHandleCompleteInboundCommand);
+
+        ExpectedTrace[] queueInitiatedTraces = {
+                rabbitTemplateConvertAndSendTrace,
+                channelNBasicPublishTrace,
+                rabbitMqConsumerInvocationTrace,
+                amqChannelHandleCompleteInboundCommandTrace
+        };
+
+        // verify client-initiated traces
+        Method rabbitTemplateReceive = rabbitTemplateClass.getDeclaredMethod("receive", String.class);
+        ExpectedTrace rabbitTemplateReceiveTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateReceive); // method
+        Method channelNBasicGet = channelNClass.getDeclaredMethod("basicGet", String.class, boolean.class);
+        ExpectedTrace channelNBasicGetTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL,
+                channelNBasicGet);
+        Class<?> propagationMarkerClass = PropagationMarker.class;
+        Method propagationMarkerMark = propagationMarkerClass.getDeclaredMethod("mark");
+        ExpectedTrace markTrace = Expectations.event(
+                ServiceType.INTERNAL_METHOD.getName(),
+                propagationMarkerMark);
+
+        ExpectedTrace[] clientInitiatedTraces = {
+                rabbitTemplateReceiveTrace,
+                channelNBasicGetTrace,
+                markTrace
+        };
+
+        int expectedTraceCount = queueInitiatedTraces.length + clientInitiatedTraces.length;
+        final PluginTestVerifier verifier = testRunner.runPull(expectedTraceCount);
+
+        verifier.verifyDiscreteTrace(queueInitiatedTraces);
+        verifier.verifyDiscreteTrace(clientInitiatedTraces);
+
+        verifier.verifyTraceCount(0);
+    }
+
+    @Test
+    public void testPullWithTimeout() throws Exception {
+        final String remoteAddress = testRunner.getRemoteAddress();
+
+        Class<?> rabbitTemplateClass = Class.forName("org.springframework.amqp.rabbit.core.RabbitTemplate");
+        // verify queue-initiated traces
+        Method rabbitTemplateConvertAndSend = rabbitTemplateClass.getDeclaredMethod("convertAndSend", String.class, String.class, Object.class);
+        ExpectedTrace rabbitTemplateConvertAndSendTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateConvertAndSend); // method
+        // automatic recovery deliberately disabled as Spring has it's own recovery mechanism
+        Class<?> channelNClass = Class.forName("com.rabbitmq.client.impl.ChannelN");
+        Method channelNBasicPublish = channelNClass.getDeclaredMethod("basicPublish", String.class, String.class, boolean.class, boolean.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace channelNBasicPublishTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                channelNBasicPublish, // method
+                null, // rpc
+                remoteAddress, // endPoint
+                "exchange-" + RabbitMQTestConstants.EXCHANGE, // destinationId
+                Expectations.annotation("rabbitmq.exchange", RabbitMQTestConstants.EXCHANGE),
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PULL));
+        ExpectedTrace rabbitMqConsumerInvocationTrace = Expectations.root(
+                RabbitMQTestConstants.RABBITMQ_CLIENT, // serviceType
+                "RabbitMQ Consumer Invocation", // method
+                "rabbitmq://exchange=" + RabbitMQTestConstants.EXCHANGE, // rpc
+                null, // endPoint (collected but API to retrieve local address is not available in all versions, so skip)
+                remoteAddress, // remoteAddress
+                Expectations.annotation("rabbitmq.routingkey", RabbitMQTestConstants.ROUTING_KEY_PULL));
+        Class<?> consumerDispatcherClass = Class.forName("com.rabbitmq.client.impl.ConsumerDispatcher");
+        Method consumerDispatcherHandleDelivery = consumerDispatcherClass.getDeclaredMethod("handleDelivery", Consumer.class, String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace consumerDispatcherHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                consumerDispatcherHandleDelivery); // method
+        ExpectedTrace asynchronousInvocationTrace = Expectations.event(
+                ServiceType.ASYNC.getName(),
+                "Asynchronous Invocation");
+        // RabbitTemplate internal consumer implementation - may change in future versions which will cause tests to
+        // fail, in which case the integration test needs to be updated to match code changes
+        Class<?> rabbitTemplateInternalConsumerClass = Class.forName("org.springframework.amqp.rabbit.core.RabbitTemplate$2");
+        Method rabbitTemplateInternalConsumerHandleDelivery = rabbitTemplateInternalConsumerClass.getDeclaredMethod("handleDelivery", String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace rabbitTemplateInternalConsumerHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateInternalConsumerHandleDelivery); // method
+        Class<?> deliveryClass = Class.forName("org.springframework.amqp.rabbit.support.Delivery");
+        Constructor<?> deliveryConstructor = deliveryClass.getDeclaredConstructor(String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class, String.class);
+        ExpectedTrace deliveryConstructorTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                deliveryConstructor);
+
+        ExpectedTrace[] queueInitiatedTraces = {
+                rabbitTemplateConvertAndSendTrace,
+                channelNBasicPublishTrace,
+                rabbitMqConsumerInvocationTrace,
+                consumerDispatcherHandleDeliveryTrace,
+                asynchronousInvocationTrace,
+                rabbitTemplateInternalConsumerHandleDeliveryTrace,
+                deliveryConstructorTrace
+        };
+
+        // verify client-initiated traces
+        Method rabbitTemplateReceive = rabbitTemplateClass.getDeclaredMethod("receive", String.class, long.class);
+        ExpectedTrace rabbitTemplateReceiveTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateReceive); // method
+        Class<?> propagationMarkerClass = PropagationMarker.class;
+        Method propagationMarkerMark = propagationMarkerClass.getDeclaredMethod("mark");
+        ExpectedTrace markTrace = Expectations.event(
+                ServiceType.INTERNAL_METHOD.getName(),
+                propagationMarkerMark);
+
+        ExpectedTrace[] clientInitiatedTraces = {
+                rabbitTemplateReceiveTrace,
+                markTrace
+        };
+
+        int expectedTraceCount = queueInitiatedTraces.length + clientInitiatedTraces.length;
+        final PluginTestVerifier verifier = testRunner.runPull(expectedTraceCount, 5000L);
+
+        verifier.verifyDiscreteTrace(queueInitiatedTraces);
+        verifier.verifyDiscreteTrace(clientInitiatedTraces);
+
+        verifier.verifyTraceCount(0);
+    }
+}