diff --git a/CHANGES.md b/CHANGES.md index 0836436292..2984b9541a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -17,6 +17,16 @@ Release Notes. error when any of them fails. A batch of a cluster connection gets one span per slot, as Lettuce writes the commands of each slot as a batch of their own. Mark the span of a reactive Lettuce command as an error when Redis replies with an error, such as `WRONGTYPE`. +* Fix the `NullPointerException`s thrown by the `rocketMQ-5.x` plugin, and the callback span stopped by the wrong + callback, when an asynchronous send with a callback succeeds on rocketmq-client 5.1.4 or later, which wraps the + callback in a `SendCallback` of its own (apache/skywalking#14189). +* Fix the missing `peer` and `mq.broker` of the consumer spans of the `rocketMQ-client-java-5.x` plugin for push + consumers built by rocketmq-client-java 5.0.8 or later, which construct `PushConsumerImpl` with another constructor. +* Stop the `rocketMQ-client-java-5.x` plugin from dropping the lite topic (rocketmq-client-java 5.1.0 or later) and the + priority (5.2.0 or later) of the messages it traces, which turned them into normal messages. The plugin rebuilt every + message to add its headers, and now adds them to the copy of the message that the client sends. +* Test the `rocketMQ-5.x` plugin with rocketmq-client 5.5, and the `rocketMQ-client-java-5.x` plugin with + rocketmq-client-java 5.0, 5.1 and 5.2. All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/267?closed=1) diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java index f18f0f8c0e..127a4f3fa0 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java @@ -34,6 +34,11 @@ /** * {@link OnSuccessInterceptor} create local span when the method {@link org.apache.rocketmq.client.producer.SendCallback#onSuccess(SendResult)} * execute. + *

+ * {@link MessageSendInterceptor} attaches the {@link SendCallBackEnhanceInfo} to the callback passed to {@link + * org.apache.rocketmq.client.impl.MQClientAPIImpl}. A callback invoked by that callback, such as the user callback that + * {@code DefaultMQProducerImpl$BackpressureSendCallBack} wraps since rocketmq-client 5.1.4, usually has none. It runs + * inside the span of the outer callback, so all three methods skip it, which keeps the span stack balanced. */ public class OnSuccessInterceptor implements InstanceMethodsAroundInterceptor { @@ -43,6 +48,9 @@ public class OnSuccessInterceptor implements InstanceMethodsAroundInterceptor { public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) throws Throwable { SendCallBackEnhanceInfo enhanceInfo = (SendCallBackEnhanceInfo) objInst.getSkyWalkingDynamicField(); + if (enhanceInfo == null) { + return; + } AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + enhanceInfo.getTopicId() + "/Producer/Callback"); activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER); SendStatus sendStatus = ((SendResult) allArguments[0]).getSendStatus(); @@ -56,6 +64,9 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { + if (objInst.getSkyWalkingDynamicField() == null) { + return ret; + } ContextManager.stopSpan(); return ret; } @@ -63,6 +74,9 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { + if (objInst.getSkyWalkingDynamicField() == null) { + return; + } ContextManager.activeSpan().log(t); } } diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageGetPropertiesInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageGetPropertiesInterceptor.java new file mode 100644 index 0000000000..ede67b5676 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageGetPropertiesInterceptor.java @@ -0,0 +1,65 @@ +/* + * 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.rocketMQ.client.java.v5; + +import java.lang.reflect.Method; +import java.util.Map; +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.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.util.StringUtil; + +/** + * {@link MessageGetPropertiesInterceptor} adds the context that {@link PublishingMessageImplInterceptor} injected into + * a {@code PublishingMessageImpl} to the new map that {@code MessageImpl#getProperties()} returns. Other messages carry + * no context and are left alone. + */ +public class MessageGetPropertiesInterceptor implements InstanceMethodsAroundInterceptor { + + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + } + + @Override + @SuppressWarnings("unchecked") + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + Object contextCarrier = objInst.getSkyWalkingDynamicField(); + if (!(contextCarrier instanceof ContextCarrier) || !(ret instanceof Map)) { + return ret; + } + Map properties = (Map) ret; + CarrierItem next = ((ContextCarrier) contextCarrier).items(); + while (next.hasNext()) { + next = next.next(); + if (!StringUtil.isEmpty(next.getHeadValue())) { + properties.put(next.getHeadKey(), next.getHeadValue()); + } + } + return ret; + } + + @Override + public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + } +} diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendAsyncInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendAsyncInterceptor.java index dcd4cc2874..9ee7f8eed5 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendAsyncInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendAsyncInterceptor.java @@ -20,16 +20,11 @@ import java.lang.reflect.Method; import java.util.Collection; -import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.apis.message.Message; -import org.apache.rocketmq.client.apis.message.MessageBuilder; import org.apache.rocketmq.client.apis.producer.SendReceipt; import org.apache.rocketmq.client.java.impl.ClientImpl; -import org.apache.rocketmq.client.java.message.MessageBuilderImpl; -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; import org.apache.skywalking.apm.agent.core.context.tag.StringTag; import org.apache.skywalking.apm.agent.core.context.tag.Tags; @@ -40,11 +35,10 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; import org.apache.skywalking.apm.agent.core.util.CollectionUtil; import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; -import org.apache.skywalking.apm.util.StringUtil; /** * {@link MessageSendAsyncInterceptor} create exit span when the method {@link org.apache.rocketmq.client.java.impl.producer.ProducerImpl#sendAsync(org.apache.rocketmq.client.apis.message.Message)} - * execute + * execute. {@link PublishingMessageImplInterceptor} injects the span into the message that is sent. */ public class MessageSendAsyncInterceptor implements InstanceMethodsAroundInterceptor { @@ -62,10 +56,8 @@ public void beforeMethod(EnhancedInstance objInst, Message message = (Message) allArguments[0]; ClientImpl producerImpl = (ClientImpl) objInst; - ContextCarrier contextCarrier = new ContextCarrier(); String namingServiceAddress = producerImpl.getClientConfiguration().getEndpoints(); - AbstractSpan span = ContextManager.createExitSpan( - buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress); + AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), namingServiceAddress); span.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER); Tags.MQ_BROKER.set(span, namingServiceAddress); Tags.MQ_TOPIC.set(span, message.getTopic()); @@ -79,38 +71,7 @@ public void beforeMethod(EnhancedInstance objInst, Optional tag = message.getTag(); tag.ifPresent(s -> span.tag(MQ_MESSAGE_TAGS, s)); } - - contextCarrier.extensionInjector().injectSendingTimestamp(); SpanLayer.asMQ(span); - - Map properties = message.getProperties(); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - if (!StringUtil.isEmpty(next.getHeadValue())) { - properties.put(next.getHeadKey(), next.getHeadValue()); - } - } - - MessageBuilder messageBuilder = new MessageBuilderImpl(); - messageBuilder.setTopic(message.getTopic()); - if (message.getTag().isPresent()) { - messageBuilder.setTag(message.getTag().get()); - } - messageBuilder.setKeys(message.getKeys().toArray(new String[0])); - if (message.getMessageGroup().isPresent()) { - messageBuilder.setMessageGroup(message.getMessageGroup().get()); - } - - byte[] body = new byte[message.getBody().limit()]; - message.getBody().get(body); - messageBuilder.setBody(body); - if (message.getDeliveryTimestamp().isPresent()) { - messageBuilder.setDeliveryTimestamp(message.getDeliveryTimestamp().get()); - } - - properties.forEach(messageBuilder::addProperty); - allArguments[0] = messageBuilder.build(); } @Override diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendInterceptor.java index e2e32bedea..54d1f1719f 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendInterceptor.java @@ -19,13 +19,9 @@ package org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5; import org.apache.rocketmq.client.apis.message.Message; -import org.apache.rocketmq.client.apis.message.MessageBuilder; import org.apache.rocketmq.client.apis.producer.SendReceipt; import org.apache.rocketmq.client.apis.producer.Transaction; import org.apache.rocketmq.client.java.impl.ClientImpl; -import org.apache.rocketmq.client.java.message.MessageBuilderImpl; -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; import org.apache.skywalking.apm.agent.core.context.tag.StringTag; import org.apache.skywalking.apm.agent.core.context.tag.Tags; @@ -36,17 +32,16 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; import org.apache.skywalking.apm.agent.core.util.CollectionUtil; import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; -import org.apache.skywalking.apm.util.StringUtil; import java.lang.reflect.Method; import java.util.Collection; -import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; /** * {@link MessageSendInterceptor} create exit span when the method {@link org.apache.rocketmq.client.java.impl.producer.ProducerImpl#send(Message)} * and {@link org.apache.rocketmq.client.java.impl.producer.ProducerImpl#send(Message, Transaction)} execute. + * {@link PublishingMessageImplInterceptor} injects the span into the message that is sent. */ public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor { @@ -60,9 +55,8 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr Message message = (Message) allArguments[0]; ClientImpl producerImpl = (ClientImpl) objInst; - ContextCarrier contextCarrier = new ContextCarrier(); String namingServiceAddress = producerImpl.getClientConfiguration().getEndpoints(); - AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress); + AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), namingServiceAddress); span.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER); Tags.MQ_BROKER.set(span, namingServiceAddress); Tags.MQ_TOPIC.set(span, message.getTopic()); @@ -78,37 +72,7 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr span.tag(MQ_MESSAGE_TAGS, tag.get()); } } - - contextCarrier.extensionInjector().injectSendingTimestamp(); SpanLayer.asMQ(span); - - Map properties = message.getProperties(); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - if (!StringUtil.isEmpty(next.getHeadValue())) { - properties.put(next.getHeadKey(), next.getHeadValue()); - } - } - - MessageBuilder messageBuilder = new MessageBuilderImpl(); - messageBuilder.setTopic(message.getTopic()); - if (message.getTag().isPresent()) { - messageBuilder.setTag(message.getTag().get()); - } - messageBuilder.setKeys(message.getKeys().toArray(new String[0])); - if (message.getMessageGroup().isPresent()) { - messageBuilder.setMessageGroup(message.getMessageGroup().get()); - } - - byte[] body = new byte[message.getBody().limit()]; - message.getBody().get(body); - messageBuilder.setBody(body); - if (message.getDeliveryTimestamp().isPresent()) { - messageBuilder.setDeliveryTimestamp(message.getDeliveryTimestamp().get()); - } - properties.entrySet().forEach(item -> messageBuilder.addProperty(item.getKey(), item.getValue())); - allArguments[0] = messageBuilder.build(); } @Override diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PublishingMessageImplInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PublishingMessageImplInterceptor.java new file mode 100644 index 0000000000..ee10e97127 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PublishingMessageImplInterceptor.java @@ -0,0 +1,52 @@ +/* + * 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.rocketMQ.client.java.v5; + +import org.apache.skywalking.apm.agent.core.context.ContextCarrier; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.NoopSpan; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; + +/** + * {@link PublishingMessageImplInterceptor} injects the exit span of the send, created by {@link + * MessageSendInterceptor} or {@link MessageSendAsyncInterceptor}, into the {@code PublishingMessageImpl} that the + * producer copies the message into. {@link MessageGetPropertiesInterceptor} adds it to the user properties of the + * message that is sent, so the message the application passed keeps every attribute, such as the lite topic and the + * priority of newer clients. When the trace is ignored, the active span is a {@link NoopSpan} instead of an exit span, + * and the carrier still carries the correlation context and the sending timestamp. + */ +public class PublishingMessageImplInterceptor implements InstanceConstructorInterceptor { + + @Override + public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { + if (!ContextManager.isActive()) { + return; + } + AbstractSpan activeSpan = ContextManager.activeSpan(); + if (!activeSpan.isExit() && !(activeSpan instanceof NoopSpan)) { + return; + } + ContextCarrier contextCarrier = new ContextCarrier(); + ContextManager.inject(contextCarrier); + contextCarrier.extensionInjector().injectSendingTimestamp(); + objInst.setSkyWalkingDynamicField(contextCarrier); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PushConsumerImplInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PushConsumerImplInterceptor.java index d6aa1ec5fa..0d72f634fd 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PushConsumerImplInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PushConsumerImplInterceptor.java @@ -24,10 +24,9 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; import org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.ConsumerEnhanceInfos; -import java.util.Map; - /** - * {@link PushConsumerImplInterceptor} create exit span when the method {@link org.apache.rocketmq.client.java.impl.consumer.PushConsumerImpl#PushConsumerImpl(ClientConfiguration, String, Map, MessageListener, int, int, int)} execute. + * {@link PushConsumerImplInterceptor} passes the endpoints of the {@link ClientConfiguration} to the {@link + * MessageListener} when a {@link org.apache.rocketmq.client.java.impl.consumer.PushConsumerImpl} is constructed. */ public class PushConsumerImplInterceptor implements InstanceConstructorInterceptor { diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/MessageImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/MessageImplInstrumentation.java new file mode 100644 index 0000000000..0346e4fdf2 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/MessageImplInstrumentation.java @@ -0,0 +1,72 @@ +/* + * 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.rocketMQ.client.java.v5.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * {@code MessageImpl#getProperties()} returns a new map, which {@code PublishingMessageImpl}, a subclass, puts into + * the user properties of the message it sends. + */ +public class MessageImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.java.message.MessageImpl"; + private static final String GET_PROPERTIES_METHOD_NAME = "getProperties"; + private static final String GET_PROPERTIES_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.MessageGetPropertiesInterceptor"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[0]; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[]{ + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(GET_PROPERTIES_METHOD_NAME).and(takesArguments(0)); + } + + @Override + public String getMethodsInterceptor() { + return GET_PROPERTIES_INTERCEPTOR; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + } + }; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplAsyncInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplAsyncInstrumentation.java index 7492d3a3ec..6beef10893 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplAsyncInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplAsyncInstrumentation.java @@ -60,7 +60,7 @@ public String getMethodsInterceptor() { @Override public boolean isOverrideArgs() { - return true; + return false; } } }; diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplInstrumentation.java index 22a8c123c2..aaeee5f291 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ProducerImplInstrumentation.java @@ -60,7 +60,7 @@ public String getMethodsInterceptor() { @Override public boolean isOverrideArgs() { - return true; + return false; } } }; diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PublishingMessageImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PublishingMessageImplInstrumentation.java new file mode 100644 index 0000000000..e182e58491 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PublishingMessageImplInstrumentation.java @@ -0,0 +1,64 @@ +/* + * 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.rocketMQ.client.java.v5.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.any; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * The producer copies every message it sends into a {@code PublishingMessageImpl}, inside the exit span of the send. + */ +public class PublishingMessageImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.java.message.PublishingMessageImpl"; + private static final String CONSTRUCTOR_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.PublishingMessageImplInterceptor"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[]{ + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return any(); + } + + @Override + public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR; + } + } + }; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[0]; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PushConsumerImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PushConsumerImplInstrumentation.java index 5b68b5acaf..15bd290aab 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PushConsumerImplInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/PushConsumerImplInstrumentation.java @@ -25,7 +25,8 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; -import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; public class PushConsumerImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { @@ -43,7 +44,11 @@ public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { new ConstructorInterceptPoint() { @Override public ElementMatcher getConstructorMatcher() { - return takesArguments(7); + // The builder calls the 7-argument constructor up to 5.0.7, the 8-argument one in 5.0.8, and + // the 9-argument one since 5.1.0. The shorter ones delegate to the longest one, so a direct + // call of a shorter one is intercepted twice and sets the same info twice. + return takesArgument(0, named("org.apache.rocketmq.client.apis.ClientConfiguration")) + .and(takesArgument(3, named("org.apache.rocketmq.client.apis.consumer.MessageListener"))); } @Override diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/resources/skywalking-plugin.def index 9c1da6962d..4257e25f6a 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/resources/skywalking-plugin.def @@ -20,3 +20,5 @@ rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.SimpleConsumerImplInstrumentation rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.ProducerImplAsyncInstrumentation rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.SimpleConsumerImplAsyncInstrumentation +rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.PublishingMessageImplInstrumentation +rocketMQ-client-java-5.x=org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.MessageImplInstrumentation diff --git a/test/plugin/scenarios/rocketmq-5-grpc-scenario/support-version.list b/test/plugin/scenarios/rocketmq-5-grpc-scenario/support-version.list index c7b3c10e09..12086b7e1d 100644 --- a/test/plugin/scenarios/rocketmq-5-grpc-scenario/support-version.list +++ b/test/plugin/scenarios/rocketmq-5-grpc-scenario/support-version.list @@ -16,4 +16,8 @@ # lists your version here (Contains only the last version number of each minor version.) -5.4.0 +# The first column is the version of the apache/rocketmq broker image, rocketmq.client.version is the version of +# rocketmq-client-java under test. +5.3.3,rocketmq.client.version=5.0.8 +5.3.4,rocketmq.client.version=5.1.0 +5.4.0,rocketmq.client.version=5.2.2 diff --git a/test/plugin/scenarios/rocketmq-scenario/support-version.list b/test/plugin/scenarios/rocketmq-scenario/support-version.list index 8dc0b9e562..73ea0c1916 100644 --- a/test/plugin/scenarios/rocketmq-scenario/support-version.list +++ b/test/plugin/scenarios/rocketmq-scenario/support-version.list @@ -16,6 +16,7 @@ # lists your version here (Contains only the last version number of each minor version.) +5.5.0 5.4.0 4.9.4 4.8.0