Fix RocketMQ latestDepTest (#6731)

Fixes #6729 
Fixes #6730
This commit is contained in:
Mateusz Rzeszutek 2022-09-23 18:16:22 +02:00 committed by GitHub
parent d26e456d1b
commit 557c6d1290
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 41 additions and 2 deletions

View File

@ -23,4 +23,6 @@ dependencies {
tasks.withType<Test>().configureEach {
jvmArgs("-Dotel.instrumentation.rocketmq-client.experimental-span-attributes=true")
systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
}

View File

@ -12,3 +12,7 @@ dependencies {
testImplementation(project(":instrumentation:rocketmq-client-4.8:testing"))
}
tasks.withType<Test>().configureEach {
systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
}

View File

@ -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() {

View File

@ -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() {}