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. - *

- * 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 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 getProperties() { - if (this.properties == null) { - if (this.msgMetadataBuilder.getPropertiesCount() > 0) { - Map internalProperties = new HashMap(); - 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. - *

- * 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. - *

- * 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 *

* Here is the intercept process steps: * * 

- *  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 messageProcessed method
  *  3. Extract all the Trace Context when call messageProcessed method
  *  4. Capture trace context and set into SkyWalkingDynamic field if consumer has a message listener when messageProcessed 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:
  *
  * 
- *  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 sendAsync method
  *  3. Inject the context to {@link Message#getProperties()}
  *  4. Create {@link SendCallbackEnhanceRequiredInfo} with ContextManager.capture() 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 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 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 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 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,