Bump mockito-core from 3.11.2 to 4.0.0

apm-agent-core/src/test/resources/json-specs/span_types.json

@@ -274,6 +274,12 @@
         "__used_by": [
           "ruby"
         ]
+      },
+      "rocketmq": {
+        "__description": "Apache RocketMQ",
+        "__used_by": [
+          "java"
+        ]
       }
     }
   },

apm-agent-plugins/apm-rocketmq-plugin/pom.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>apm-agent-plugins</artifactId>
+        <groupId>co.elastic.apm</groupId>
+        <version>1.26.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>apm-rocketmq-plugin</artifactId>
+
+    <properties>
+        <rocketmq.client.version>4.2.0</rocketmq.client.version>
+        <apm-agent-parent.base.dir>${project.basedir}/../..</apm-agent-parent.base.dir>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>${rocketmq.client.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-broker</artifactId>
+            <version>4.2.0</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <artifactId>logback-classic</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>logback-core</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-namesrv</artifactId>
+            <version>4.2.0</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>logback-classic</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>logback-core</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+    </dependencies>
+</project>

apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/BaseRocketmqInstrumentation.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.rocketmq;
+
+import co.elastic.apm.agent.bci.TracerAwareInstrumentation;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public abstract class BaseRocketmqInstrumentation extends TracerAwareInstrumentation {
+
+    @Override
+    public Collection<String> getInstrumentationGroupNames() {
+        return Collections.singletonList("rocketmq");
+    }
+}

apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/Labels.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.rocketmq;
+
+public class Labels {
+
+    public static final String KEYS = "rocketmq.keys";
+
+    public static final String MSG_ID = "rocketmq.msgId";
+
+    public static final String NAME_SRV = "rocketmq.namesrv";
+
+    public static final String SEND_STATUS = "rocketmq.sendStatus";
+}

apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConcurrentListenerInstrumentation.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.rocketmq.consumer;
+
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.List;
+
+import static net.bytebuddy.matcher.ElementMatchers.*;
+
+public class ConcurrentListenerInstrumentation extends BaseRocketmqInstrumentation {
+
+    @Override
+    public ElementMatcher<? super TypeDescription> getTypeMatcher() {
+        return hasSuperType(named("org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently"))
+            .and(not(isInterface()))
+            .and(not(isAbstract()));
+    }
+
+    @Override
+    public ElementMatcher<? super MethodDescription> getMethodMatcher() {
+        return named("consumeMessage");
+    }
+
+    public static class AdviceClass {
+
+        @Advice.OnMethodEnter(inline = false, suppress = Throwable.class)
+        public static Object onEnter(@Advice.Argument(0) List<MessageExt> msgList,
+                                   @Advice.This Object thiz) {
+            return ConsumerHelper.get().onConsumeStart(thiz, msgList);
+        }
+
+        @Advice.OnMethodExit(inline = false, suppress = Throwable.class, onThrowable = Throwable.class)
+        public static void onExit(@Advice.Thrown Throwable thrown,
+                                  @Advice.Return ConsumeConcurrentlyStatus status,
+                                  @Advice.Enter Object transactionObj) {
+
+            if (transactionObj == null) {
+                return;
+            }
+            Transaction transaction = (Transaction) transactionObj;
+            ConsumerHelper helper = ConsumerHelper.get();
+            if (thrown != null) {
+                helper.onConsumeError(transaction, thrown);
+            } else {
+                helper.onConsumeEnd(transaction, status.name(), status == ConsumeConcurrentlyStatus.CONSUME_SUCCESS);
+            }
+        }
+    }
+}

apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConsumerHelper.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.rocketmq.consumer;
+
+import co.elastic.apm.agent.impl.ElasticApmTracer;
+import co.elastic.apm.agent.impl.GlobalTracer;
+import co.elastic.apm.agent.impl.transaction.Outcome;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.rocketmq.Labels;
+import co.elastic.apm.agent.rocketmq.namesrv.NameSrvHolder;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.List;
+
+public class ConsumerHelper {
+
+    private static final ConsumerHelper INSTANCE = new ConsumerHelper();
+
+    private final ElasticApmTracer tracer = GlobalTracer.requireTracerImpl();
+
+    public static ConsumerHelper get() {
+        return INSTANCE;
+    }
+
+    public Transaction onConsumeStart(Object listener, List<MessageExt> msgList) {
+        if (tracer.getActive() != null) {
+            return null;
+        }
+
+        Transaction transaction = tracer.startRootTransaction(MessageExt.class.getClassLoader());
+        if (transaction == null) {
+            return null;
+        }
+
+        // msgList.size() equals to DefaultMQPushConsumer.consumeMessageBatchMaxSize
+        // By default, DefaultMQPushConsumer.consumeMessageBatchMaxSize = 1, and rarely changed
+        MessageExt msg = msgList.get(0);
+        transaction.withType("messaging")
+            .withName("RocketMQ Consume ")
+            .appendToName(msg.getTopic()).appendToName("#").appendToName(msg.getTags())
+            .addLabel(Labels.MSG_ID, msg.getMsgId());
+        if (msg.getKeys() != null) {
+            transaction.addLabel(Labels.KEYS, msg.getKeys());
+        }
+
+        DefaultMQPushConsumer consumer = PushConsumerHolder.get(listener);
+        if (consumer != null) {
+            transaction.addLabel(Labels.NAME_SRV, NameSrvHolder.get(consumer, consumer, PushConsumerNameSrvGetter.get()));
+        }
+
+        return transaction.activate();
+    }
+
+    public void onConsumeEnd(Transaction transaction, String consumeStatus, boolean isSuccess) {
+        transaction.withResult(consumeStatus)
+            .withOutcome(isSuccess ? Outcome.SUCCESS : Outcome.FAILURE)
+            .deactivate().end();
+    }
+
+    public void onConsumeError(Transaction transaction, Throwable thrown) {
+        transaction.captureException(thrown).deactivate().end();
+    }
+}

apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/OrderListenerInstrumentation.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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 co.elastic.apm.agent.rocketmq.consumer;
+
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.List;
+
+import static net.bytebuddy.matcher.ElementMatchers.*;
+
+public class OrderListenerInstrumentation extends BaseRocketmqInstrumentation {
+
+    @Override
+    public ElementMatcher<? super TypeDescription> getTypeMatcher() {
+        return hasSuperType(named("org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly"))
+            .and(not(isInterface()))
+            .and(not(isAbstract()));
+    }
+
+    @Override
+    public ElementMatcher<? super MethodDescription> getMethodMatcher() {
+        return named("consumeMessage");
+    }
+
+    public static class AdviceClass {
+
+        @Advice.OnMethodEnter(inline = false, suppress = Throwable.class)
+        public static Object onEnter(@Advice.Argument(0) List<MessageExt> msgList,
+                                   @Advice.This Object thiz) {
+            return ConsumerHelper.get().onConsumeStart(thiz, msgList);
+        }
+
+        @Advice.OnMethodExit(inline = false, suppress = Throwable.class, onThrowable = Throwable.class)
+        public static void onExit(@Advice.Thrown Throwable thrown,
+                                  @Advice.Return ConsumeOrderlyStatus status,
+                                  @Advice.Enter Object transactionObj) {
+            if (transactionObj == null) {
+                return;
+            }
+
+            Transaction transaction = (Transaction) transactionObj;
+            ConsumerHelper helper = ConsumerHelper.get();
+            if (thrown != null) {
+                helper.onConsumeError(transaction, thrown);
+            } else {
+                helper.onConsumeEnd(transaction, status.name(), status == ConsumeOrderlyStatus.SUCCESS);
+            }
+        }
+    }
+}