From f17a4ea54a4245fef776ed66e102465e72a97ed4 Mon Sep 17 00:00:00 2001
From: "pg.yang" <pg.yang@hotmail.com>
Date: Sat, 5 Nov 2022 22:18:32 +0800
Subject: [PATCH] Polish up Pulsar plugins (#375)

---
 CHANGES.md                                    |  1 +
 .../ProducerConstructorInterceptor.java       | 21 +---
 .../apm/plugin/pulsar/MockMessage.java        | 82 ----------------
 .../ProducerConstructorInterceptorTest.java   | 96 ------------------
 .../v28x/ProducerConstructorInterceptor.java  | 21 +---
 .../ProducerConstructorInterceptorTest.java   | 98 -------------------
 .../ConsumerConstructorInterceptor.java       | 20 +---
 .../common/ConsumerEnhanceRequiredInfo.java   | 65 ------------
 .../common/ProducerEnhanceRequiredInfo.java   | 65 ------------
 .../common/PulsarConsumerInterceptor.java     | 34 ++++---
 .../common/PulsarProducerInterceptor.java     | 18 ++--
 .../ConsumerConstructorInterceptorTest.java   | 88 -----------------
 .../pulsar/common/MockConsumerImpl.java       | 57 +++++++++++
 .../pulsar/common/MockProducerImpl.java       | 48 +++++++++
 .../common/PulsarConsumerInterceptorTest.java | 51 +++++-----
 ...PulsarConsumerListenerInterceptorTest.java | 52 +++++-----
 .../common/PulsarProducerInterceptorTest.java | 34 +++----
 .../pulsar-scenario/config/expectedData.yaml  | 32 +++---
 18 files changed, 223 insertions(+), 660 deletions(-)
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/MockMessage.java
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptorTest.java
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptorTest.java
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerEnhanceRequiredInfo.java
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ProducerEnhanceRequiredInfo.java
 delete mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptorTest.java
 create mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockConsumerImpl.java
 create mode 100644 apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockProducerImpl.java

diff --git a/CHANGES.md b/CHANGES.md
index 8580366fa..412ab5bed 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -26,6 +26,7 @@ Release Notes.
 * Add RocketMQ test scenarios for version 4.3 - 4.9. No 4.0 - 4.2 release images for testing.
 * Support mannual propagation of tracing context to next operators for webflux.
 * Add MQ_TOPIC and MQ_BROKER tags for RocketMQ consumer's span. 
+* Polish up Pulsar plugins to remove unnecessary dynamic value , set peer at consumer side 
 
 #### Documentation
 
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptor.java
index 924f7c5d7..78bc0a43a 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptor.java
@@ -18,32 +18,13 @@
 
 package org.apache.skywalking.apm.plugin.pulsar;
 
-import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
-import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
 
-/**
- * Interceptor of pulsar producer constructor.
- * <p>
- * The interceptor create {@link ProducerEnhanceRequiredInfo} which is required by instance method interceptor, So use
- * it to update the skywalking dynamic field of pulsar producer enhanced instance. So that the instance methods can get
- * the {@link ProducerEnhanceRequiredInfo}
- */
 public class ProducerConstructorInterceptor implements InstanceConstructorInterceptor {
 
     @Override
     public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
-        PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
-        String topic = (String) allArguments[1];
-        ProducerEnhanceRequiredInfo producerEnhanceRequiredInfo = new ProducerEnhanceRequiredInfo();
-        producerEnhanceRequiredInfo.setTopic(topic);
-        /*
-         * Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
-         * can handle the service url provider which use a dynamic service url
-         */
-        producerEnhanceRequiredInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
-        producerEnhanceRequiredInfo.setPropertiesInjector(new MessagePropertiesInjectorV27());
-        objInst.setSkyWalkingDynamicField(producerEnhanceRequiredInfo);
+        objInst.setSkyWalkingDynamicField(new MessagePropertiesInjectorV27());
     }
 }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/MockMessage.java b/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/MockMessage.java
deleted file mode 100644
index f70e5a3b6..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/MockMessage.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar;
-
-import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.impl.MessageImpl;
-import org.apache.pulsar.common.api.proto.PulsarApi;
-import org.apache.pulsar.shade.io.netty.buffer.ByteBuf;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-public class MockMessage extends MessageImpl implements EnhancedInstance {
-
-    private PulsarApi.MessageMetadata.Builder msgMetadataBuilder = PulsarApi.MessageMetadata.newBuilder();
-
-    private transient Map<String, String> properties;
-
-    private Object enhancedSkyWalkingField;
-
-    public MockMessage() {
-        this(null, "1:1", new HashMap(), null, null);
-    }
-
-    public MockMessage(String topic, String msgId, Map properties, ByteBuf payload, Schema schema) {
-        super(topic, msgId, properties, payload, schema);
-    }
-
-    @Override
-    public PulsarApi.MessageMetadata.Builder getMessageBuilder() {
-        return msgMetadataBuilder;
-    }
-
-    public synchronized Map<String, String> getProperties() {
-        if (this.properties == null) {
-            if (this.msgMetadataBuilder.getPropertiesCount() > 0) {
-                Map<String, String> internalProperties = new HashMap<String, String>();
-                for (int i = 0; i < this.msgMetadataBuilder.getPropertiesCount(); i++) {
-                    PulsarApi.KeyValue kv = this.msgMetadataBuilder.getProperties(i);
-                    internalProperties.put(kv.getKey(), kv.getValue());
-                }
-                this.properties = Collections.unmodifiableMap(internalProperties);
-            } else {
-                this.properties = Collections.emptyMap();
-            }
-        }
-        return this.properties;
-    }
-
-    @Override
-    public String getProperty(String name) {
-        return this.getProperties().get(name);
-    }
-
-    @Override
-    public Object getSkyWalkingDynamicField() {
-        return enhancedSkyWalkingField;
-    }
-
-    @Override
-    public void setSkyWalkingDynamicField(Object value) {
-        this.enhancedSkyWalkingField = value;
-    }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptorTest.java
deleted file mode 100644
index 71f700305..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptorTest.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
-import static org.mockito.Mockito.when;
-
-import org.apache.pulsar.client.impl.LookupService;
-import org.apache.pulsar.client.impl.MessageImpl;
-import org.apache.pulsar.client.impl.PulsarClientImpl;
-import org.apache.skywalking.apm.agent.core.context.CarrierItem;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
-import org.apache.skywalking.apm.plugin.pulsar.common.MessagePropertiesInjector;
-import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-
-@RunWith(MockitoJUnitRunner.class)
-public class ProducerConstructorInterceptorTest {
-
-    private static final String SERVICE_URL = "pulsar://localhost:6650";
-    private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
-    private static final String HEAD_KEY = "testKey";
-    private static final String HEAD_VALUE = "testValue";
-
-    @Mock
-    private PulsarClientImpl pulsarClient;
-    @Mock
-    private LookupService lookupService;
-    @Mock
-    private CarrierItem carrierItem;
-    private MessageImpl<?> message;
-
-    private ProducerConstructorInterceptor constructorInterceptor;
-
-    private EnhancedInstance enhancedInstance = new EnhancedInstance() {
-
-        private ProducerEnhanceRequiredInfo requiredInfo;
-
-        @Override
-        public Object getSkyWalkingDynamicField() {
-            return requiredInfo;
-        }
-
-        @Override
-        public void setSkyWalkingDynamicField(Object value) {
-            this.requiredInfo = (ProducerEnhanceRequiredInfo) value;
-        }
-    };
-
-    @Before
-    public void setUp() {
-        when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
-        when(pulsarClient.getLookup()).thenReturn(lookupService);
-        when(carrierItem.getHeadKey()).thenReturn(HEAD_KEY);
-        when(carrierItem.getHeadValue()).thenReturn(HEAD_VALUE);
-        constructorInterceptor = new ProducerConstructorInterceptor();
-        message = new MockMessage();
-    }
-
-    @Test
-    public void testOnConsumer() {
-        constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
-            pulsarClient,
-            TOPIC_NAME
-        });
-        ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
-        assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
-        assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
-        final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
-        Assert.assertNotNull(propertiesInjector);
-        propertiesInjector.inject(message, carrierItem);
-        assertThat(message.getProperty(HEAD_KEY), is(HEAD_VALUE));
-    }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptor.java
index ce3098da2..fc199027b 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptor.java
@@ -18,32 +18,13 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.v28x;
 
-import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
-import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
 
-/**
- * Interceptor of pulsar producer constructor.
- * <p>
- * The interceptor create {@link ProducerEnhanceRequiredInfo} which is required by instance method interceptor, So use
- * it to update the skywalking dynamic field of pulsar producer enhanced instance. So that the instance methods can get
- * the {@link ProducerEnhanceRequiredInfo}
- */
 public class ProducerConstructorInterceptor implements InstanceConstructorInterceptor {
 
     @Override
     public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
-        PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
-        String topic = (String) allArguments[1];
-        ProducerEnhanceRequiredInfo producerEnhanceRequiredInfo = new ProducerEnhanceRequiredInfo();
-        producerEnhanceRequiredInfo.setTopic(topic);
-        /*
-         * Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
-         * can handle the service url provider which use a dynamic service url
-         */
-        producerEnhanceRequiredInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
-        producerEnhanceRequiredInfo.setPropertiesInjector(new MessagePropertiesInjectorV28());
-        objInst.setSkyWalkingDynamicField(producerEnhanceRequiredInfo);
+        objInst.setSkyWalkingDynamicField(new MessagePropertiesInjectorV28());
     }
 }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptorTest.java
deleted file mode 100644
index 2884c31f1..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptorTest.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar.v28x;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
-import static org.mockito.Mockito.when;
-
-import java.nio.ByteBuffer;
-import org.apache.pulsar.client.impl.LookupService;
-import org.apache.pulsar.client.impl.MessageImpl;
-import org.apache.pulsar.client.impl.PulsarClientImpl;
-import org.apache.pulsar.common.api.proto.MessageMetadata;
-import org.apache.skywalking.apm.agent.core.context.CarrierItem;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
-import org.apache.skywalking.apm.plugin.pulsar.common.MessagePropertiesInjector;
-import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-
-@RunWith(MockitoJUnitRunner.class)
-public class ProducerConstructorInterceptorTest {
-
-    private static final String SERVICE_URL = "pulsar://localhost:6650";
-    private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
-    private static final String HEAD_KEY = "testKey";
-    private static final String HEAD_VALUE = "testValue";
-
-    @Mock
-    private PulsarClientImpl pulsarClient;
-    @Mock
-    private LookupService lookupService;
-    @Mock
-    private CarrierItem carrierItem;
-    private MessageImpl<?> message;
-
-    private ProducerConstructorInterceptor constructorInterceptor;
-
-    private EnhancedInstance enhancedInstance = new EnhancedInstance() {
-
-        private ProducerEnhanceRequiredInfo requiredInfo;
-
-        @Override
-        public Object getSkyWalkingDynamicField() {
-            return requiredInfo;
-        }
-
-        @Override
-        public void setSkyWalkingDynamicField(Object value) {
-            this.requiredInfo = (ProducerEnhanceRequiredInfo) value;
-        }
-    };
-
-    @Before
-    public void setUp() {
-        when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
-        when(pulsarClient.getLookup()).thenReturn(lookupService);
-        when(carrierItem.getHeadKey()).thenReturn(HEAD_KEY);
-        when(carrierItem.getHeadValue()).thenReturn(HEAD_VALUE);
-        constructorInterceptor = new ProducerConstructorInterceptor();
-        message = MessageImpl.create(new MessageMetadata(), ByteBuffer.allocate(1), null);
-    }
-
-    @Test
-    public void testOnConsumer() {
-        constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
-                pulsarClient,
-                TOPIC_NAME
-        });
-        ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
-        assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
-        assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
-        final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
-        Assert.assertNotNull(propertiesInjector);
-        propertiesInjector.inject(message, carrierItem);
-        assertThat(message.getProperty(HEAD_KEY), is(HEAD_VALUE));
-    }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptor.java
index f1df30d45..d343a5da8 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptor.java
@@ -18,33 +18,15 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.common;
 
-import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
 
-/**
- * Interceptor of pulsar consumer constructor.
- * <p>
- * The interceptor create {@link ConsumerEnhanceRequiredInfo} which is required by instance method interceptor, So use
- * it to update the skywalking dynamic field of pulsar consumer enhanced instance. So that the instance methods can get
- * the {@link ConsumerEnhanceRequiredInfo}
- */
 public class ConsumerConstructorInterceptor implements InstanceConstructorInterceptor {
 
     @Override
     public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
-        PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
-        String topic = (String) allArguments[1];
         ConsumerConfigurationData consumerConfigurationData = (ConsumerConfigurationData) allArguments[2];
-        ConsumerEnhanceRequiredInfo requireInfo = new ConsumerEnhanceRequiredInfo();
-        /*
-         * Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
-         * can handle the service url provider which use a dynamic service url
-         */
-        requireInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
-        requireInfo.setTopic(topic);
-        requireInfo.setSubscriptionName(consumerConfigurationData.getSubscriptionName());
-        objInst.setSkyWalkingDynamicField(requireInfo);
+        objInst.setSkyWalkingDynamicField(consumerConfigurationData.getSubscriptionName());
     }
 }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerEnhanceRequiredInfo.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerEnhanceRequiredInfo.java
deleted file mode 100644
index 1d99828ad..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerEnhanceRequiredInfo.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar.common;
-
-/**
- * Pulsar consumer enhance required info is required by consumer enhanced object method interceptor
- */
-public class ConsumerEnhanceRequiredInfo {
-
-    /**
-     * service url of the consumer
-     */
-    private String serviceUrl;
-
-    /**
-     * topic name of the consumer
-     */
-    private String topic;
-
-    /**
-     * subscription name of the consumer
-     */
-    private String subscriptionName;
-
-    public String getServiceUrl() {
-        return serviceUrl;
-    }
-
-    public void setServiceUrl(String serviceUrl) {
-        this.serviceUrl = serviceUrl;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
-        this.topic = topic;
-    }
-
-    public String getSubscriptionName() {
-        return subscriptionName;
-    }
-
-    public void setSubscriptionName(String subscriptionName) {
-        this.subscriptionName = subscriptionName;
-    }
-
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ProducerEnhanceRequiredInfo.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ProducerEnhanceRequiredInfo.java
deleted file mode 100644
index 6f14601ba..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ProducerEnhanceRequiredInfo.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar.common;
-
-/**
- * Pulsar producer enhance required info is required by producer enhanced object method interceptor
- */
-public class ProducerEnhanceRequiredInfo {
-
-    /**
-     * service url of the pulsar producer
-     */
-    private String serviceUrl;
-
-    /**
-     * topic name of the pulsar producer
-     */
-    private String topic;
-
-    /**
-     * Message properties injector
-     */
-    private MessagePropertiesInjector propertiesInjector;
-
-    public MessagePropertiesInjector getPropertiesInjector() {
-        return propertiesInjector;
-    }
-
-    public void setPropertiesInjector(
-            MessagePropertiesInjector propertiesInjector) {
-        this.propertiesInjector = propertiesInjector;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public void setTopic(String topic) {
-        this.topic = topic;
-    }
-
-    public String getServiceUrl() {
-        return serviceUrl;
-    }
-
-    public void setServiceUrl(String serviceUrl) {
-        this.serviceUrl = serviceUrl;
-    }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptor.java
index 068e13bda..d442f8426 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptor.java
@@ -18,7 +18,9 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.common;
 
+import java.lang.reflect.Method;
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.impl.ConsumerImpl;
 import org.apache.skywalking.apm.agent.core.context.CarrierItem;
 import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
 import org.apache.skywalking.apm.agent.core.context.ContextManager;
@@ -30,15 +32,13 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceM
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 
-import java.lang.reflect.Method;
-
 /**
  * Interceptor for pulsar consumer enhanced instance
  * <p>
  * Here is the intercept process steps:
  *
  * <pre>
- *  1. Get the @{@link ConsumerEnhanceRequiredInfo} and record the service url, topic name and subscription name
+ *  1. Record the service url, topic name and subscription name through this(ConsumerImpl)
  *  2. Create the entry span when call <code>messageProcessed</code> method
  *  3. Extract all the <code>Trace Context</code> when call <code>messageProcessed</code> method
  *  4. Capture trace context and set into SkyWalkingDynamic field if consumer has a message listener when <code>messageProcessed</code> method finished
@@ -52,9 +52,11 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
 
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
-        MethodInterceptResult result) throws Throwable {
+                             MethodInterceptResult result) throws Throwable {
         if (allArguments[0] != null) {
-            ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField();
+            ConsumerImpl consumer = (ConsumerImpl) objInst;
+            final String serviceUrl = consumer.getClient().getLookup().getServiceUrl();
+
             Message msg = (Message) allArguments[0];
             ContextCarrier carrier = new ContextCarrier();
             CarrierItem next = carrier.items();
@@ -62,29 +64,31 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
                 next = next.next();
                 next.setHeadValue(msg.getProperty(next.getHeadKey()));
             }
-            AbstractSpan activeSpan = ContextManager.createEntrySpan(OPERATE_NAME_PREFIX + requiredInfo.getTopic() + CONSUMER_OPERATE_NAME + requiredInfo
-                .getSubscriptionName(), carrier);
+            AbstractSpan activeSpan = ContextManager.createEntrySpan(
+                OPERATE_NAME_PREFIX + consumer.getTopic() + CONSUMER_OPERATE_NAME + objInst.getSkyWalkingDynamicField(),
+                carrier
+            );
             activeSpan.setComponent(ComponentsDefine.PULSAR_CONSUMER);
             SpanLayer.asMQ(activeSpan);
-            Tags.MQ_BROKER.set(activeSpan, requiredInfo.getServiceUrl());
-            Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopic());
+            Tags.MQ_BROKER.set(activeSpan, serviceUrl);
+            Tags.MQ_TOPIC.set(activeSpan, consumer.getTopic());
+            activeSpan.setPeer(serviceUrl);
         }
     }
 
     @Override
     public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
-        Object ret) throws Throwable {
+                              Object ret) throws Throwable {
         if (allArguments[0] != null) {
-            final ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst
-                    .getSkyWalkingDynamicField();
+            ConsumerImpl consumer = (ConsumerImpl) objInst;
             EnhancedInstance msg = (EnhancedInstance) allArguments[0];
             MessageEnhanceRequiredInfo messageEnhanceRequiredInfo = (MessageEnhanceRequiredInfo) msg
-                    .getSkyWalkingDynamicField();
+                .getSkyWalkingDynamicField();
             if (messageEnhanceRequiredInfo == null) {
                 messageEnhanceRequiredInfo = new MessageEnhanceRequiredInfo();
                 msg.setSkyWalkingDynamicField(messageEnhanceRequiredInfo);
             }
-            messageEnhanceRequiredInfo.setTopic(requiredInfo.getTopic());
+            messageEnhanceRequiredInfo.setTopic(consumer.getTopic());
             messageEnhanceRequiredInfo.setContextSnapshot(ContextManager.capture());
             ContextManager.stopSpan();
         }
@@ -93,7 +97,7 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
 
     @Override
     public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
-        Class<?>[] argumentsTypes, Throwable t) {
+                                      Class<?>[] argumentsTypes, Throwable t) {
         if (allArguments[0] != null) {
             ContextManager.activeSpan().log(t);
         }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptor.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptor.java
index eb0ca83de..070963279 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptor.java
@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.plugin.pulsar.common;
 
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.skywalking.apm.agent.core.context.CarrierItem;
 import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
 import org.apache.skywalking.apm.agent.core.context.ContextManager;
@@ -40,7 +41,7 @@ import java.lang.reflect.Method;
  * Here is the intercept process steps:
  *
  * <pre>
- *  1. Get the {@link ProducerEnhanceRequiredInfo} and record the service url, topic name
+ *  1. Record the service url, topic name through this(ProducerImpl)
  *  2. Create the exit span when the producer invoke <code>sendAsync</code> method
  *  3. Inject the context to {@link Message#getProperties()}
  *  4. Create {@link SendCallbackEnhanceRequiredInfo} with <code>ContextManager.capture()</code> and set the
@@ -57,19 +58,18 @@ public class PulsarProducerInterceptor implements InstanceMethodsAroundIntercept
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
         MethodInterceptResult result) throws Throwable {
         if (allArguments[0] != null) {
-            ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField();
             ContextCarrier contextCarrier = new ContextCarrier();
-            String topicName = requiredInfo.getTopic();
-            AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + topicName + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, requiredInfo
-                .getServiceUrl());
-            Tags.MQ_BROKER.set(activeSpan, requiredInfo.getServiceUrl());
-            Tags.MQ_TOPIC.set(activeSpan, topicName);
+            ProducerImpl producer = (ProducerImpl) objInst;
+            final String serviceUrl = producer.getClient().getLookup().getServiceUrl();
+            AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + producer.getTopic() + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, serviceUrl);
+            Tags.MQ_BROKER.set(activeSpan, serviceUrl);
+            Tags.MQ_TOPIC.set(activeSpan, producer.getTopic());
             contextCarrier.extensionInjector().injectSendingTimestamp();
             SpanLayer.asMQ(activeSpan);
             activeSpan.setComponent(ComponentsDefine.PULSAR_PRODUCER);
             CarrierItem next = contextCarrier.items();
             MessageImpl msg = (MessageImpl) allArguments[0];
-            final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
+            MessagePropertiesInjector propertiesInjector = (MessagePropertiesInjector) objInst.getSkyWalkingDynamicField();
             if (propertiesInjector != null) {
                 while (next.hasNext()) {
                     next = next.next();
@@ -83,7 +83,7 @@ public class PulsarProducerInterceptor implements InstanceMethodsAroundIntercept
                     ContextSnapshot snapshot = ContextManager.capture();
                     if (null != snapshot) {
                         SendCallbackEnhanceRequiredInfo callbackRequiredInfo = new SendCallbackEnhanceRequiredInfo();
-                        callbackRequiredInfo.setTopic(topicName);
+                        callbackRequiredInfo.setTopic(producer.getTopic());
                         callbackRequiredInfo.setContextSnapshot(snapshot);
                         callbackInstance.setSkyWalkingDynamicField(callbackRequiredInfo);
                     }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptorTest.java
deleted file mode 100644
index a77a1b1fc..000000000
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptorTest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.pulsar.common;
-
-import org.apache.pulsar.client.impl.LookupService;
-import org.apache.pulsar.client.impl.PulsarClientImpl;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class ConsumerConstructorInterceptorTest {
-
-    private static final String SERVICE_URL = "pulsar://localhost:6650";
-    private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
-    private static final String SUBSCRIPTION_NAME = "my-sub";
-
-    @Mock
-    private PulsarClientImpl pulsarClient;
-
-    @Mock
-    private LookupService lookupService;
-
-    @Mock
-    private ConsumerConfigurationData consumerConfigurationData;
-
-    private ConsumerConstructorInterceptor constructorInterceptor;
-
-    private EnhancedInstance enhancedInstance = new EnhancedInstance() {
-
-        private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
-
-        @Override
-        public Object getSkyWalkingDynamicField() {
-            return consumerEnhanceRequiredInfo;
-        }
-
-        @Override
-        public void setSkyWalkingDynamicField(Object value) {
-            consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo) value;
-        }
-    };
-
-    @Before
-    public void setUp() {
-        when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
-        when(pulsarClient.getLookup()).thenReturn(lookupService);
-        when(consumerConfigurationData.getSubscriptionName()).thenReturn(SUBSCRIPTION_NAME);
-        constructorInterceptor = new ConsumerConstructorInterceptor();
-    }
-
-    @Test
-    public void testOnConsumer() {
-        constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
-            pulsarClient,
-            TOPIC_NAME,
-            consumerConfigurationData
-        });
-        ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
-        assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
-        assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
-        assertThat(requiredInfo.getSubscriptionName(), is(SUBSCRIPTION_NAME));
-    }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockConsumerImpl.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockConsumerImpl.java
new file mode 100644
index 000000000..f25150acb
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockConsumerImpl.java
@@ -0,0 +1,57 @@
+/*
+ *   Licensed to the Apache Software Foundation (ASF) under one or more
+ *   contributor license agreements.  See the NOTICE file distributed with
+ *   this work for additional information regarding copyright ownership.
+ *   The ASF licenses this file to You under the Apache License, Version 2.0
+ *   (the "License"); you may not use this file except in compliance with
+ *   the License.  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package org.apache.skywalking.apm.plugin.pulsar.common;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.ConsumerInterceptors;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+
+public class MockConsumerImpl extends ConsumerImpl implements EnhancedInstance {
+
+    protected MockConsumerImpl(final PulsarClientImpl client,
+                               final String topic,
+                               final ConsumerConfigurationData conf,
+                               final ExecutorService listenerExecutor,
+                               final int partitionIndex,
+                               final CompletableFuture subscribeFuture,
+                               final MessageId startMessageId,
+                               final Schema schema,
+                               final ConsumerInterceptors interceptors,
+                               final long backoffIntervalNanos,
+                               final long maxBackoffIntervalNanos) {
+        super(client, topic, conf, listenerExecutor, partitionIndex, subscribeFuture, null, startMessageId,
+              schema, interceptors, backoffIntervalNanos, maxBackoffIntervalNanos
+        );
+    }
+
+    @Override
+    public Object getSkyWalkingDynamicField() {
+        return null;
+    }
+
+    @Override
+    public void setSkyWalkingDynamicField(final Object value) {
+
+    }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockProducerImpl.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockProducerImpl.java
new file mode 100644
index 000000000..4af2c1249
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockProducerImpl.java
@@ -0,0 +1,48 @@
+/*
+ *   Licensed to the Apache Software Foundation (ASF) under one or more
+ *   contributor license agreements.  See the NOTICE file distributed with
+ *   this work for additional information regarding copyright ownership.
+ *   The ASF licenses this file to You under the Apache License, Version 2.0
+ *   (the "License"); you may not use this file except in compliance with
+ *   the License.  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package org.apache.skywalking.apm.plugin.pulsar.common;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.ProducerImpl;
+import org.apache.pulsar.client.impl.ProducerInterceptors;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+
+public class MockProducerImpl extends ProducerImpl implements EnhancedInstance {
+    public MockProducerImpl(final PulsarClientImpl client,
+                            final String topic,
+                            final ProducerConfigurationData conf,
+                            final CompletableFuture producerCreatedFuture,
+                            final int partitionIndex,
+                            final Schema schema,
+                            final ProducerInterceptors interceptors) {
+        super(client, topic, conf, producerCreatedFuture, partitionIndex, schema, interceptors);
+    }
+
+    @Override
+    public Object getSkyWalkingDynamicField() {
+        return null;
+    }
+
+    @Override
+    public void setSkyWalkingDynamicField(final Object value) {
+
+    }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptorTest.java
index c58681784..0013d1537 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptorTest.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptorTest.java
@@ -18,6 +18,9 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.common;
 
+import java.util.List;
+import org.apache.pulsar.client.impl.LookupService;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
@@ -37,11 +40,10 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 
-import java.util.List;
-
 import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_CONSUMER;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertNotNull;
@@ -57,45 +59,32 @@ public class PulsarConsumerInterceptorTest {
     @Rule
     public AgentServiceRule serviceRule = new AgentServiceRule();
 
-    private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
-
-    private MessageEnhanceRequiredInfo messageEnhanceRequiredInfo;
-
-    private PulsarConsumerInterceptor consumerInterceptor;
-
     private MockMessage msg;
 
     private EnhancedInstance consumerInstance = new EnhancedInstance() {
         @Override
         public Object getSkyWalkingDynamicField() {
-            return consumerEnhanceRequiredInfo;
+            return "my-sub";
         }
 
         @Override
         public void setSkyWalkingDynamicField(Object value) {
-            consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo) value;
+
         }
     };
 
     @Before
     public void setUp() {
-        consumerInterceptor = new PulsarConsumerInterceptor();
-        consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
-
-        consumerEnhanceRequiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
-        consumerEnhanceRequiredInfo.setServiceUrl("pulsar://localhost:6650");
-        consumerEnhanceRequiredInfo.setSubscriptionName("my-sub");
         msg = new MockMessage();
         msg.getMessageBuilder()
                 .addProperties(PulsarApi.KeyValue.newBuilder()
                         .setKey(SW8CarrierItem.HEADER_NAME)
                         .setValue("1-My40LjU=-MS4yLjM=-3-c2VydmljZQ==-aW5zdGFuY2U=-L2FwcA==-MTI3LjAuMC4xOjgwODA="));
-        messageEnhanceRequiredInfo = new MessageEnhanceRequiredInfo();
-        msg.setSkyWalkingDynamicField(messageEnhanceRequiredInfo);
     }
 
     @Test
     public void testConsumerWithNullMessage() throws Throwable {
+        PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
         consumerInterceptor.beforeMethod(consumerInstance, null, new Object[] {null}, new Class[0], null);
         consumerInterceptor.afterMethod(consumerInstance, null, new Object[] {null}, new Class[0], null);
 
@@ -105,8 +94,10 @@ public class PulsarConsumerInterceptorTest {
 
     @Test
     public void testConsumerWithMessage() throws Throwable {
-        consumerInterceptor.beforeMethod(consumerInstance, null, new Object[] {msg}, new Class[0], null);
-        consumerInterceptor.afterMethod(consumerInstance, null, new Object[] {msg}, new Class[0], null);
+        EnhancedInstance enhancedInstance = mockConsumer();
+        PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
+        consumerInterceptor.beforeMethod(enhancedInstance, null, new Object[] {msg}, new Class[0], null);
+        consumerInterceptor.afterMethod(enhancedInstance, null, new Object[] {msg}, new Class[0], null);
 
         List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
         assertThat(traceSegments.size(), is(1));
@@ -122,8 +113,10 @@ public class PulsarConsumerInterceptorTest {
 
     @Test
     public void testConsumerWithMessageListener() throws Throwable {
-        consumerInterceptor.beforeMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
-        consumerInterceptor.afterMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
+        EnhancedInstance enhancedInstance = mockConsumer();
+        PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
+        consumerInterceptor.beforeMethod(enhancedInstance, null, new Object[]{msg}, new Class[0], null);
+        consumerInterceptor.afterMethod(enhancedInstance, null, new Object[]{msg}, new Class[0], null);
 
         List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
         assertThat(traceSegments.size(), is(1));
@@ -137,8 +130,6 @@ public class PulsarConsumerInterceptorTest {
         assertConsumerSpan(spans.get(0));
 
         final MessageEnhanceRequiredInfo requiredInfo = (MessageEnhanceRequiredInfo) msg.getSkyWalkingDynamicField();
-        assertThat(requiredInfo.getTopic(), is(
-                ((ConsumerEnhanceRequiredInfo) consumerInstance.getSkyWalkingDynamicField()).getTopic()));
         assertNotNull(requiredInfo.getContextSnapshot());
     }
 
@@ -155,4 +146,16 @@ public class PulsarConsumerInterceptorTest {
         MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3));
         MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("3.4.5"));
     }
+
+    private EnhancedInstance mockConsumer() throws Throwable {
+        EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockConsumerImpl.class);
+        final LookupService lookup = PowerMockito.mock(LookupService.class);
+        final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
+        PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
+        PowerMockito.when(client, "getLookup").thenReturn(lookup);
+        PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
+        PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
+        PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn("my-sub");
+        return pulsarProducerInstance;
+    }
 }
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerListenerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerListenerInterceptorTest.java
index ceb806db1..b84db19df 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerListenerInterceptorTest.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerListenerInterceptorTest.java
@@ -18,7 +18,11 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.common;
 
+import java.util.List;
+import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.impl.LookupService;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
@@ -39,11 +43,10 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 
-import java.util.List;
-
 import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_CONSUMER;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertNotNull;
@@ -53,7 +56,6 @@ import static org.junit.Assert.assertThat;
 @RunWith(PowerMockRunner.class)
 @PowerMockRunnerDelegate(TracingSegmentRunner.class)
 public class PulsarConsumerListenerInterceptorTest {
-
     @Rule
     public AgentServiceRule serviceRule = new AgentServiceRule();
     @SegmentStoragePoint
@@ -70,32 +72,13 @@ public class PulsarConsumerListenerInterceptorTest {
     };
     private PulsarConsumerListenerInterceptor consumerListenerInterceptor;
     private MockMessage msg;
-    private PulsarConsumerInterceptor consumerInterceptor;
-    private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
-    private final EnhancedInstance consumerInstance = new EnhancedInstance() {
-        @Override
-        public Object getSkyWalkingDynamicField() {
-            return consumerEnhanceRequiredInfo;
-        }
-
-        @Override
-        public void setSkyWalkingDynamicField(Object value) {
-            consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo) value;
-        }
-    };
     private MockConsumer consumer;
     private MessageListener messageListener;
 
     @Before
     public void setUp() {
-        consumerInterceptor = new PulsarConsumerInterceptor();
         consumerListenerInterceptor = new PulsarConsumerListenerInterceptor();
         messageListener = (consumer, message) -> message.getTopicName();
-        consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
-
-        consumerEnhanceRequiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
-        consumerEnhanceRequiredInfo.setServiceUrl("pulsar://localhost:6650");
-        consumerEnhanceRequiredInfo.setSubscriptionName("my-sub");
         msg = new MockMessage();
         msg.getMessageBuilder()
                 .addProperties(PulsarApi.KeyValue.newBuilder()
@@ -134,16 +117,17 @@ public class PulsarConsumerListenerInterceptorTest {
 
     @Test
     public void testWithMessageListenerHasRequiredInfo() throws Throwable {
-        consumerInterceptor.beforeMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
-        consumerInterceptor.afterMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
+        PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
+        EnhancedInstance consumer =  mockConsumer();
+        consumerInterceptor.beforeMethod(consumer, null, new Object[]{msg}, new Class[0], null);
+        consumerInterceptor.afterMethod(consumer, null, new Object[]{msg}, new Class[0], null);
         consumerListenerInterceptor
                 .beforeMethod(consumerConfigurationDataInstance, null, new Object[0], new Class[0], null);
         final MessageListener enhancedMessageListener = (MessageListener) consumerListenerInterceptor
                 .afterMethod(consumerConfigurationDataInstance, null, new Object[0], new Class[0],
                         this.messageListener);
         assertNotNull(enhancedMessageListener);
-        enhancedMessageListener.received(consumer, msg);
-
+        enhancedMessageListener.received((Consumer) consumer, msg);
         List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
         assertThat(traceSegments.size(), is(2));
 
@@ -156,6 +140,22 @@ public class PulsarConsumerListenerInterceptorTest {
         assertConsumerSpan(spans.get(0));
     }
 
+    private EnhancedInstance mockConsumer() {
+        EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockConsumerImpl.class);
+        final LookupService lookup = PowerMockito.mock(LookupService.class);
+        final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
+        try {
+            PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
+            PowerMockito.when(client, "getLookup").thenReturn(lookup);
+            PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
+            PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
+            PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn("my-sub");
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        return pulsarProducerInstance;
+    }
+
     private void assertConsumerSpan(AbstractTracingSpan span) {
         SpanAssert.assertLayer(span, SpanLayer.MQ);
         SpanAssert.assertComponent(span, PULSAR_CONSUMER);
diff --git a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptorTest.java
index 9f42347a9..16b26be87 100644
--- a/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptorTest.java
+++ b/apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptorTest.java
@@ -18,7 +18,10 @@
 
 package org.apache.skywalking.apm.plugin.pulsar.common;
 
+import java.util.List;
+import org.apache.pulsar.client.impl.LookupService;
 import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
 import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
@@ -33,11 +36,10 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 
-import java.util.List;
-
 import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_PRODUCER;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -57,21 +59,6 @@ public class PulsarProducerInterceptorTest {
     private Object[] arguments;
     private Class[] argumentType;
 
-    private EnhancedInstance pulsarProducerInstance = new EnhancedInstance() {
-
-        @Override
-        public Object getSkyWalkingDynamicField() {
-            ProducerEnhanceRequiredInfo requiredInfo = new ProducerEnhanceRequiredInfo();
-            requiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
-            requiredInfo.setServiceUrl("pulsar://localhost:6650");
-            return requiredInfo;
-        }
-
-        @Override
-        public void setSkyWalkingDynamicField(Object value) {
-        }
-    };
-
     private MessageImpl msg = new MockMessage();
 
     @Before
@@ -86,6 +73,16 @@ public class PulsarProducerInterceptorTest {
 
     @Test
     public void testSendMessage() throws Throwable {
+
+        EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockProducerImpl.class);
+        final LookupService lookup = PowerMockito.mock(LookupService.class);
+        final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
+        PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
+        PowerMockito.when(client, "getLookup").thenReturn(lookup);
+        PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
+        PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
+        PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn((MessagePropertiesInjector) (message, carrierItem) -> {
+            });
         producerInterceptor.beforeMethod(pulsarProducerInstance, null, arguments, argumentType, null);
         producerInterceptor.afterMethod(pulsarProducerInstance, null, arguments, argumentType, null);
 
@@ -101,6 +98,9 @@ public class PulsarProducerInterceptorTest {
 
     @Test
     public void testSendWithNullMessage() throws Throwable {
+        EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockProducerImpl.class);
+        PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn((MessagePropertiesInjector) (message, carrierItem) -> {
+        });
         producerInterceptor.beforeMethod(pulsarProducerInstance, null, new Object[] {null}, argumentType, null);
         producerInterceptor.afterMethod(pulsarProducerInstance, null, new Object[] {null}, argumentType, null);
         List<TraceSegment> traceSegmentList = segmentStorage.getTraceSegments();
diff --git a/test/plugin/scenarios/pulsar-scenario/config/expectedData.yaml b/test/plugin/scenarios/pulsar-scenario/config/expectedData.yaml
index 57000dc63..8b675a1b0 100644
--- a/test/plugin/scenarios/pulsar-scenario/config/expectedData.yaml
+++ b/test/plugin/scenarios/pulsar-scenario/config/expectedData.yaml
@@ -28,9 +28,9 @@ segmentItems:
           componentId: 73
           isError: false
           spanType: Exit
-          peer: not null
+          peer: not blank
           tags:
-            - { key: mq.broker, value: not null }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: test }
           skipAnalysis: 'false'
         - operationName: Pulsar/testMultiPartition/Producer
@@ -42,10 +42,10 @@ segmentItems:
           componentId: 73
           isError: false
           spanType: Exit
-          peer: not null
+          peer: not blank
           skipAnalysis: false
           tags:
-            - { key: mq.broker, value: not null }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: testMultiPartition }
         - operationName: GET:/case/pulsar-case
           parentSpanId: -1
@@ -92,10 +92,10 @@ segmentItems:
           componentId: 74
           isError: false
           spanType: Entry
-          peer: ''
+          peer: not blank
           tags:
-            - { key: transmission.latency, value: not null }
-            - { key: mq.broker, value: not null }
+            - { key: transmission.latency, value: not blank }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: test }
           refs:
             - { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null, refType: CrossProcess,
@@ -113,11 +113,11 @@ segmentItems:
           componentId: 74
           isError: false
           spanType: Entry
-          peer: ''
+          peer: not blank
           skipAnalysis: 'false'
           tags:
-            - { key: transmission.latency, value: not null }
-            - { key: mq.broker, value: not null }
+            - { key: transmission.latency, value: not blank }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: test }
           refs:
             - { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null,
@@ -173,10 +173,10 @@ segmentItems:
           componentId: 74
           isError: false
           spanType: Entry
-          peer: ''
+          peer: not blank
           tags:
-            - { key: transmission.latency, value: not null }
-            - { key: mq.broker, value: not null }
+            - { key: transmission.latency, value: not blank }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: testMultiPartition }
           refs:
             - { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null, refType: CrossProcess,
@@ -194,11 +194,11 @@ segmentItems:
           componentId: 74
           isError: false
           spanType: Entry
-          peer: ''
+          peer: not blank
           skipAnalysis: 'false'
           tags:
-            - { key: transmission.latency, value: not null }
-            - { key: mq.broker, value: not null }
+            - { key: transmission.latency, value: not blank }
+            - { key: mq.broker, value: not blank }
             - { key: mq.topic, value: testMultiPartition }
           refs:
             - { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null,