From 557c6d12907274395e2ad7fa3736e44d50d51175 Mon Sep 17 00:00:00 2001
From: Mateusz Rzeszutek <mrzeszutek@splunk.com>
Date: Fri, 23 Sep 2022 18:16:22 +0200
Subject: [PATCH] Fix RocketMQ latestDepTest (#6731)

Fixes #6729
Fixes #6730
---
 .../javaagent/build.gradle.kts                |  2 ++
 .../library/build.gradle.kts                  |  4 +++
 .../AbstractRocketMqClientTest.groovy         |  5 +++
 .../main/java/base/IntegrationTestBase.java   | 32 +++++++++++++++++--
 4 files changed, 41 insertions(+), 2 deletions(-)

diff --git a/instrumentation/rocketmq-client-4.8/javaagent/build.gradle.kts b/instrumentation/rocketmq-client-4.8/javaagent/build.gradle.kts
index 7ebeca1c3e..b5a7dd46a8 100644
--- a/instrumentation/rocketmq-client-4.8/javaagent/build.gradle.kts
+++ b/instrumentation/rocketmq-client-4.8/javaagent/build.gradle.kts
@@ -23,4 +23,6 @@ dependencies {
 
 tasks.withType<Test>().configureEach {
   jvmArgs("-Dotel.instrumentation.rocketmq-client.experimental-span-attributes=true")
+
+  systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
 }
diff --git a/instrumentation/rocketmq-client-4.8/library/build.gradle.kts b/instrumentation/rocketmq-client-4.8/library/build.gradle.kts
index 35e337f381..b0a60b388c 100644
--- a/instrumentation/rocketmq-client-4.8/library/build.gradle.kts
+++ b/instrumentation/rocketmq-client-4.8/library/build.gradle.kts
@@ -12,3 +12,7 @@ dependencies {
 
   testImplementation(project(":instrumentation:rocketmq-client-4.8:testing"))
 }
+
+tasks.withType<Test>().configureEach {
+  systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
+}
diff --git a/instrumentation/rocketmq-client-4.8/testing/src/main/groovy/io/opentelemetry/instrumentation/rocketmq/AbstractRocketMqClientTest.groovy b/instrumentation/rocketmq-client-4.8/testing/src/main/groovy/io/opentelemetry/instrumentation/rocketmq/AbstractRocketMqClientTest.groovy
index 3c9c9fb719..0eda9a5aa0 100644
--- a/instrumentation/rocketmq-client-4.8/testing/src/main/groovy/io/opentelemetry/instrumentation/rocketmq/AbstractRocketMqClientTest.groovy
+++ b/instrumentation/rocketmq-client-4.8/testing/src/main/groovy/io/opentelemetry/instrumentation/rocketmq/AbstractRocketMqClientTest.groovy
@@ -61,6 +61,11 @@ abstract class AbstractRocketMqClientTest extends InstrumentationSpecification {
     configureMQProducer(producer)
     consumer = BaseConf.getConsumer(BaseConf.nsAddr, sharedTopic, "*", tracingMessageListener)
     configureMQPushConsumer(consumer)
+
+    // for RocketMQ 5.x wait a bit to ensure that consumer is properly started up
+    if (Boolean.getBoolean("testLatestDeps")) {
+      Thread.sleep(30_000)
+    }
   }
 
   def cleanupSpec() {
diff --git a/instrumentation/rocketmq-client-4.8/testing/src/main/java/base/IntegrationTestBase.java b/instrumentation/rocketmq-client-4.8/testing/src/main/java/base/IntegrationTestBase.java
index 53107a3979..2ffe40b1b6 100644
--- a/instrumentation/rocketmq-client-4.8/testing/src/main/java/base/IntegrationTestBase.java
+++ b/instrumentation/rocketmq-client-4.8/testing/src/main/java/base/IntegrationTestBase.java
@@ -5,14 +5,19 @@
 
 package base;
 
+import static java.util.Collections.emptyMap;
+
 import io.opentelemetry.instrumentation.test.utils.PortUtils;
 import java.io.File;
 import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -23,7 +28,6 @@ import org.apache.rocketmq.namesrv.NamesrvController;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyServerConfig;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
-import org.apache.rocketmq.test.util.MQAdmin;
 import org.junit.Assert;
 
 public final class IntegrationTestBase {
@@ -128,7 +132,31 @@ public final class IntegrationTestBase {
   }
 
   public static void initTopic(String topic, String nsAddr, String clusterName) {
-    MQAdmin.createTopic(nsAddr, clusterName, topic, 20);
+    try {
+      // RocketMQ 4.x
+      Class<?> mqAdmin = Class.forName("org.apache.rocketmq.test.util.MQAdmin");
+      Method createTopic =
+          mqAdmin.getMethod("createTopic", String.class, String.class, String.class, int.class);
+      createTopic.invoke(null, nsAddr, clusterName, topic, 20);
+    } catch (ClassNotFoundException
+        | InvocationTargetException
+        | NoSuchMethodException
+        | IllegalAccessException e) {
+
+      // RocketMQ 5.x
+      try {
+        Class<?> mqAdmin = Class.forName("org.apache.rocketmq.test.util.MQAdminTestUtils");
+        Method createTopic =
+            mqAdmin.getMethod(
+                "createTopic", String.class, String.class, String.class, int.class, Map.class);
+        createTopic.invoke(null, nsAddr, clusterName, topic, 20, emptyMap());
+      } catch (ClassNotFoundException
+          | InvocationTargetException
+          | NoSuchMethodException
+          | IllegalAccessException ex) {
+        throw new LinkageError("Could not initialize topic", ex);
+      }
+    }
   }
 
   private IntegrationTestBase() {}