Repository: pinpoint-apm/pinpoint
Issue: Add support for Spring amqp rabbit 1.7.7+ and 2.0.3+
Labels: enhancement
module:plugin

Spring amqp rabbit 1.7.7 and 2.0.3 have been released and introduced **BlockingQueueConsumer$ConsumerDecorator** which wraps **BlockingQueueConsumer$InternalConsumer**.
**BlockingQueueConsumer$ConsumerDecorator** also needs to be transformed to handle async trace propagation.

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/rabbitmq/README.md b/plugins/rabbitmq/README.md
index 09d7df921660..95f0c651795c 100644
--- a/plugins/rabbitmq/README.md
+++ b/plugins/rabbitmq/README.md
@@ -36,4 +36,5 @@ The plugin has been tested on the following libraries.
 ```
 com.rabbitmq:amqp-client - 2.7.0+
 org.springframework.amqp:spring-rabbit - 1.3.3+
-```
\ No newline at end of file
+```
+*(Pinpoint 1.7.2 supports up to org.springframework.amqp:spring-rabbit 1.3.3 ~ 1.7.2, 2.0.1 and 2.0.2)*
\ No newline at end of file
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 2e009167c9eb..d773d4fe46fd 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
@@ -402,6 +402,17 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader loader, Strin
                     return null;
                 }
             });
+            // Spring-rabbit 1.7.7+, 2.0.3+
+            transformTemplate.transform("org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$ConsumerDecorator", new TransformCallback() {
+                @Override
+                public byte[] doInTransform(Instrumentor instrumentor, ClassLoader loader, String className, Class<?> classBeingRedefined, ProtectionDomain protectionDomain, byte[] classfileBuffer) throws InstrumentException {
+                    InstrumentClass target = instrumentor.getInstrumentClass(loader, className, classfileBuffer);
+                    if (addConsumerHandleDeliveryInterceptor(target)) {
+                        return target.toBytecode();
+                    }
+                    return null;
+                }
+            });
         }
     }
 


Test patch:
diff --git a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_2_0_0_IT.java b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_1_7_7_IT.java
similarity index 99%
rename from agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_2_0_0_IT.java
rename to agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_1_7_7_IT.java
index c8cf2e558ba5..d6a0d9457d82 100644
--- a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_2_0_0_IT.java
+++ b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_0_to_1_7_7_IT.java
@@ -49,9 +49,9 @@
  */
 @RunWith(PinpointPluginTestSuite.class)
 @PinpointConfig("rabbitmq/client/pinpoint-rabbitmq.config")
-@Dependency({"org.springframework.amqp:spring-rabbit:[1.7.0.RELEASE,2.0.0.RELEASE)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
+@Dependency({"org.springframework.amqp:spring-rabbit:[1.7.0.RELEASE,1.7.7.RELEASE)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
 @JvmArgument({"-Dpinpoint.configcenter=false"})
-public class SpringAmqpRabbit_1_7_0_to_2_0_0_IT {
+public class SpringAmqpRabbit_1_7_0_to_1_7_7_IT {
 
     private static final TestBroker BROKER = new TestBroker();
     private static final TestApplicationContext CONTEXT = new TestApplicationContext();
diff --git a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java
new file mode 100644
index 000000000000..59671849e408
--- /dev/null
+++ b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_1_7_7_to_2_0_0_IT.java
@@ -0,0 +1,311 @@
+/*
+ * 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.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.JvmArgument;
+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;
+
+/**
+ * <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
+ */
+@RunWith(PinpointPluginTestSuite.class)
+@PinpointConfig("rabbitmq/client/pinpoint-rabbitmq.config")
+@Dependency({"org.springframework.amqp:spring-rabbit:[1.7.7.RELEASE,2.0.0.RELEASE)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
+@JvmArgument({"-Dpinpoint.configcenter=false"})
+public class SpringAmqpRabbit_1_7_7_to_2_0_0_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<?> blockingQueueConsumerConsumerDecoratorClass = Class.forName("org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$ConsumerDecorator");
+        Method blockingQueueConsumerConsumerDecoratorHandleDelivery = blockingQueueConsumerConsumerDecoratorClass.getDeclaredMethod("handleDelivery", String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace blockingQueueConsumerConsumerDecoratorHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                blockingQueueConsumerConsumerDecoratorHandleDelivery);
+        Class<?> deliveryClass = Class.forName("org.springframework.amqp.rabbit.support.Delivery");
+        Constructor<?> deliveryConstructor = deliveryClass.getDeclaredConstructor(String.class, Envelope.class, AMQP.BasicProperties.class, byte[].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,
+                blockingQueueConsumerConsumerDecoratorHandleDeliveryTrace,
+                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");
+        Class<?> queueingConsumerClass = Class.forName("com.rabbitmq.client.QueueingConsumer");
+        Method queueingConsumerHandleDelivery = queueingConsumerClass.getDeclaredMethod("handleDelivery", String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace queueingConsumerHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                queueingConsumerHandleDelivery);
+        Class<?> queueingConsumerDeliveryClass = Class.forName("com.rabbitmq.client.QueueingConsumer$Delivery");
+        Constructor queueingConsumerDeliveryConstructor = queueingConsumerDeliveryClass.getDeclaredConstructor(Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace queueingConsumerDeliveryConstructorTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL,
+                queueingConsumerDeliveryConstructor);
+
+        ExpectedTrace[] queueInitiatedTraces = {
+                rabbitTemplateConvertAndSendTrace,
+                channelNBasicPublishTrace,
+                rabbitMqConsumerInvocationTrace,
+                consumerDispatcherHandleDeliveryTrace,
+                asynchronousInvocationTrace,
+                queueingConsumerHandleDeliveryTrace,
+                queueingConsumerDeliveryConstructorTrace
+        };
+
+        // verify client-initiated traces
+        Method rabbitTemplateReceive = rabbitTemplateClass.getDeclaredMethod("receive", String.class, long.class);
+        ExpectedTrace rabbitTemplateReceiveTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                rabbitTemplateReceive); // method
+        Method queueingConsumerNextDelivery = queueingConsumerClass.getDeclaredMethod("nextDelivery", long.class);
+        ExpectedTrace queueingConsumerNextDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL,
+                queueingConsumerNextDelivery);
+        Class<?> propagationMarkerClass = PropagationMarker.class;
+        Method propagationMarkerMark = propagationMarkerClass.getDeclaredMethod("mark");
+        ExpectedTrace markTrace = Expectations.event(
+                ServiceType.INTERNAL_METHOD.getName(),
+                propagationMarkerMark);
+
+        ExpectedTrace[] clientInitiatedTraces = {
+                rabbitTemplateReceiveTrace,
+                queueingConsumerNextDeliveryTrace,
+                markTrace
+        };
+
+        int expectedTraceCount = queueInitiatedTraces.length + clientInitiatedTraces.length;
+        final PluginTestVerifier verifier = testRunner.runPull(expectedTraceCount, 5000L);
+
+        verifier.verifyDiscreteTrace(queueInitiatedTraces);
+        verifier.verifyDiscreteTrace(clientInitiatedTraces);
+
+        verifier.verifyTraceCount(0);
+    }
+}
diff --git a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_x_IT.java b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_0_to_2_0_3_IT.java
similarity index 98%
rename from agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_x_IT.java
rename to agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_0_to_2_0_3_IT.java
index e4666733cb60..d0454261c2b4 100644
--- a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_x_IT.java
+++ b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_0_to_2_0_3_IT.java
@@ -58,10 +58,10 @@
  */
 @RunWith(PinpointPluginTestSuite.class)
 @PinpointConfig("rabbitmq/client/pinpoint-rabbitmq.config")
-@Dependency({"org.springframework.amqp:spring-rabbit:[2.0.0.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.0.RELEASE,2.0.3.RELEASE)", "com.fasterxml.jackson.core:jackson-core:2.8.11", "org.apache.qpid:qpid-broker:6.1.1"})
 @JvmVersion(8)
 @JvmArgument({"-Dpinpoint.configcenter=false"})
-public class SpringAmqpRabbit_2_x_IT {
+public class SpringAmqpRabbit_2_0_0_to_2_0_3_IT {
 
     private static final TestBroker BROKER = new TestBroker();
     private static final TestApplicationContext CONTEXT = new TestApplicationContext();
@@ -306,7 +306,6 @@ public void testPullWithTimeout() throws Exception {
 
         int expectedTraceCount = queueInitiatedTraces.length + clientInitiatedTraces.length;
         final PluginTestVerifier verifier = testRunner.runPull(expectedTraceCount, 5000L);
-//        final PluginTestVerifier verifier = testRunner.runPull(9, 5000L);
 
         verifier.verifyDiscreteTrace(queueInitiatedTraces);
         verifier.verifyDiscreteTrace(clientInitiatedTraces);
diff --git a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java
new file mode 100644
index 000000000000..b9c9c067bbfe
--- /dev/null
+++ b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/rabbitmq/spring/SpringAmqpRabbit_2_0_3_to_2_x_IT.java
@@ -0,0 +1,314 @@
+/*
+ * 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.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.JvmArgument;
+import com.navercorp.pinpoint.test.plugin.JvmVersion;
+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.0.3 introduced <tt>BlockingQueueConsumer$ConsumerDecorator</tt>, which wraps
+ * <tt>BlockingQueueConsumer$InternalConsumer</tt>.
+ *
+ * <p>This affects asynchronous trace propagation as async trace object now gets attached to
+ * <tt>BlockingQueueConsumer$ConsumerDecorator</tt> instead of the previous
+ * <tt>BlockingQueueConsumer$InternalConsumer</tt>.
+ *
+ * @author HyunGil Jeong
+ */
+@RunWith(PinpointPluginTestSuite.class)
+@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"})
+@JvmVersion(8)
+@JvmArgument({"-Dpinpoint.configcenter=false"})
+public class SpringAmqpRabbit_2_0_3_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<?> blockingQueueConsumerConsumerDecoratorClass = Class.forName("org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$ConsumerDecorator");
+        Method blockingQueueConsumerConsumerDecoratorHandleDelivery = blockingQueueConsumerConsumerDecoratorClass.getDeclaredMethod("handleDelivery", String.class, Envelope.class, AMQP.BasicProperties.class, byte[].class);
+        ExpectedTrace blockingQueueConsumerConsumerDecoratorHandleDeliveryTrace = Expectations.event(
+                RabbitMQTestConstants.RABBITMQ_CLIENT_INTERNAL, // serviceType
+                blockingQueueConsumerConsumerDecoratorHandleDelivery);
+        Class<?> deliveryClass = Class.forName("org.springframework.amqp.rabbit.support.Delivery");
+        Constructor<?> deliveryConstructor = deliveryClass.getDeclaredConstructor(String.class, Envelope.class, AMQP.BasicProperties.class, byte[].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,
+                blockingQueueConsumerConsumerDecoratorHandleDeliveryTrace,
+                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);
+        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);
+    }
+}