Skip to content

Commit e588d1e

Browse files
Cirilla-zmhCirilla-zmh
committed
fix test for api and kafka-clients
1 parent 5686199 commit e588d1e

File tree

5 files changed

+15
-84
lines changed

5 files changed

+15
-84
lines changed

‎messaging-wrappers/api/build.gradle.kts‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,12 @@ otelJava.moduleName.set("io.opentelemetry.contrib.messaging.wrappers")
99

1010
dependencies {
1111
api("io.opentelemetry.instrumentation:opentelemetry-instrumentation-api-incubator")
12+
api("io.opentelemetry.semconv:opentelemetry-semconv")
13+
api("io.opentelemetry.semconv:opentelemetry-semconv-incubating")
1214

1315
compileOnly("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi")
1416
compileOnly("io.opentelemetry:opentelemetry-api-incubator")
1517

16-
implementation("io.opentelemetry.semconv:opentelemetry-semconv")
17-
implementation("io.opentelemetry.semconv:opentelemetry-semconv-incubating")
18-
1918
testImplementation("com.google.guava:guava:33.4.8-jre")
2019
testImplementation(project(":messaging-wrappers:testing"))
2120
}

‎messaging-wrappers/api/src/test/java/io/opentelemetry/contrib/messaging/wrappers/UserDefinedMessageSystemTest.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import com.google.common.eventbus.EventBus;
44
import io.opentelemetry.api.GlobalOpenTelemetry;
55
import io.opentelemetry.api.OpenTelemetry;
6+
import io.opentelemetry.api.common.AttributeKey;
67
import io.opentelemetry.api.trace.Span;
78
import io.opentelemetry.api.trace.SpanKind;
89
import io.opentelemetry.api.trace.Tracer;
@@ -26,7 +27,6 @@
2627
import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_BODY;
2728
import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_ID;
2829
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
29-
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_CLIENT_ID;
3030
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
3131
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
3232
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
@@ -102,7 +102,9 @@ public void assertTraces() {
102102
equalTo(
103103
MESSAGING_MESSAGE_BODY_SIZE,
104104
MESSAGE_BODY.getBytes(StandardCharsets.UTF_8).length),
105-
equalTo(MESSAGING_CLIENT_ID, CLIENT_ID),
105+
// FIXME: We do have "messaging.client_id" in instrumentation but "messaging.client.id" in
106+
// semconv library right now. It should be replaced after semconv release.
107+
equalTo(AttributeKey.stringKey("messaging.client_id"), CLIENT_ID),
106108
equalTo(MESSAGING_OPERATION, "process")),
107109
span ->
108110
span.hasName("process child")

‎messaging-wrappers/kafka-clients/build.gradle.kts‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -18,22 +18,12 @@ dependencies {
1818

1919
testImplementation("org.apache.kafka:kafka-clients:0.11.0.0")
2020
testImplementation("io.opentelemetry.instrumentation:opentelemetry-kafka-clients-2.6")
21+
testImplementation(project(":messaging-wrappers:testing"))
2122

2223
testAnnotationProcessor("com.google.auto.service:auto-service")
2324
testCompileOnly("com.google.auto.service:auto-service-annotations")
24-
testImplementation("org.junit.jupiter:junit-jupiter-api")
25-
testImplementation("org.junit.jupiter:junit-jupiter-params")
2625
testImplementation("org.testcontainers:kafka")
2726
testImplementation("org.testcontainers:junit-jupiter")
28-
29-
testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
30-
testImplementation("io.opentelemetry:opentelemetry-sdk")
31-
testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
32-
testImplementation("io.opentelemetry:opentelemetry-sdk-extension-incubator")
33-
testImplementation("io.opentelemetry.semconv:opentelemetry-semconv")
34-
testImplementation("io.opentelemetry.semconv:opentelemetry-semconv-incubating")
35-
testImplementation("io.opentelemetry:opentelemetry-exporter-logging")
36-
testImplementation("io.opentelemetry.javaagent:opentelemetry-testing-common")
3727
}
3828

3929
tasks {

‎messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientBaseTest.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package io.opentelemetry.contrib.messaging.wrappers.kafka;
22

3-
import io.opentelemetry.api.common.AttributeKey;
43
import java.time.Duration;
54
import java.util.Collection;
65
import java.util.Collections;
@@ -10,6 +9,8 @@
109
import java.util.concurrent.ExecutionException;
1110
import java.util.concurrent.TimeUnit;
1211
import java.util.concurrent.TimeoutException;
12+
13+
import io.opentelemetry.contrib.messaging.wrappers.testing.AbstractBaseTest;
1314
import org.apache.kafka.clients.admin.AdminClient;
1415
import org.apache.kafka.clients.admin.NewTopic;
1516
import org.apache.kafka.clients.consumer.Consumer;
@@ -37,13 +38,11 @@
3738
* */
3839
@SuppressWarnings("OtelInternalJavadoc")
3940
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
40-
public abstract class KafkaClientBaseTest {
41+
public abstract class KafkaClientBaseTest extends AbstractBaseTest {
4142

4243
private static final Logger logger = LoggerFactory.getLogger(KafkaClientBaseTest.class);
4344

4445
protected static final String SHARED_TOPIC = "shared.topic";
45-
protected static final AttributeKey<String> MESSAGING_CLIENT_ID =
46-
AttributeKey.stringKey("messaging.client_id");
4746

4847
private KafkaContainer kafka;
4948
protected Producer<Integer, String> producer;

‎messaging-wrappers/kafka-clients/src/test/java/io/opentelemetry/contrib/messaging/wrappers/kafka/KafkaClientTest.java‎

Lines changed: 5 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -2,43 +2,28 @@
22

33
import io.opentelemetry.api.GlobalOpenTelemetry;
44
import io.opentelemetry.api.OpenTelemetry;
5+
import io.opentelemetry.api.common.AttributeKey;
56
import io.opentelemetry.api.trace.Span;
67
import io.opentelemetry.api.trace.SpanKind;
78
import io.opentelemetry.api.trace.Tracer;
89
import io.opentelemetry.context.Scope;
910
import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapper;
10-
import io.opentelemetry.contrib.messaging.wrappers.kafka.internal.AutoConfiguredDataCapture;
1111
import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaConsumerAttributesExtractor;
1212
import io.opentelemetry.contrib.messaging.wrappers.kafka.semconv.KafkaProcessRequest;
1313
import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingConsumerInterceptor;
1414
import io.opentelemetry.instrumentation.kafkaclients.v2_6.TracingProducerInterceptor;
15-
import io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil;
16-
import io.opentelemetry.sdk.testing.assertj.TraceAssert;
17-
import io.opentelemetry.sdk.testing.assertj.TracesAssert;
18-
import io.opentelemetry.sdk.trace.data.SpanData;
1915
import org.apache.kafka.clients.consumer.ConsumerConfig;
2016
import org.apache.kafka.clients.consumer.ConsumerRecord;
2117
import org.apache.kafka.clients.consumer.ConsumerRecords;
2218
import org.apache.kafka.clients.producer.ProducerConfig;
2319
import org.apache.kafka.clients.producer.ProducerRecord;
2420
import org.assertj.core.api.AbstractAssert;
25-
import org.awaitility.core.ConditionTimeoutException;
2621
import org.junit.jupiter.api.Test;
2722

28-
import javax.annotation.Nullable;
2923
import java.nio.charset.StandardCharsets;
3024
import java.time.Duration;
31-
import java.util.ArrayList;
32-
import java.util.Comparator;
33-
import java.util.List;
3425
import java.util.Map;
35-
import java.util.Arrays;
36-
import java.util.concurrent.TimeoutException;
37-
import java.util.function.Consumer;
38-
import java.util.function.Supplier;
3926

40-
import static io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil.orderByRootSpanName;
41-
import static io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil.waitForTraces;
4227
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
4328
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
4429
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
@@ -49,7 +34,6 @@
4934
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
5035
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
5136
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
52-
import static org.awaitility.Awaitility.await;
5337

5438
public class KafkaClientTest extends KafkaClientBaseTest {
5539

@@ -130,7 +114,7 @@ public void consumeWithChild(Tracer tracer, MessagingProcessWrapper<KafkaProcess
130114
@SuppressWarnings("deprecation") // using deprecated semconv
131115
public void assertTraces() {
132116
waitAndAssertTraces(
133-
orderByRootSpanName("parent", "producer callback"),
117+
sortByRootSpanName("parent", "producer callback"),
134118
trace ->
135119
trace.hasSpansSatisfyingExactly(
136120
span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
@@ -152,9 +136,9 @@ public void assertTraces() {
152136
satisfies(
153137
MESSAGING_DESTINATION_PARTITION_ID,
154138
org.assertj.core.api.AbstractStringAssert::isNotEmpty),
155-
satisfies(
156-
MESSAGING_CLIENT_ID,
157-
stringAssert -> stringAssert.isEqualTo("test-consumer-1")),
139+
// FIXME: We do have "messaging.client_id" in instrumentation but "messaging.client.id" in
140+
// semconv library right now. It should be replaced after semconv release.
141+
equalTo(AttributeKey.stringKey("messaging.client_id"), "test-consumer-1"),
158142
satisfies(
159143
MESSAGING_KAFKA_OFFSET,
160144
AbstractAssert::isNotNull),
@@ -171,47 +155,4 @@ public void assertTraces() {
171155
span ->
172156
span.hasName("producer callback").hasKind(SpanKind.INTERNAL).hasNoParent()));
173157
}
174-
175-
@SafeVarargs
176-
@SuppressWarnings("varargs")
177-
private static void waitAndAssertTraces(
178-
@Nullable Comparator<List<SpanData>> traceComparator,
179-
Consumer<TraceAssert>... assertions) {
180-
List<Consumer<TraceAssert>> assertionsList = new ArrayList<>(Arrays.asList(assertions));
181-
try {
182-
await()
183-
.untilAsserted(() -> doAssertTraces(traceComparator, AutoConfiguredDataCapture::getSpans, assertionsList));
184-
} catch (Throwable t) {
185-
// awaitility is doing a jmx call that is not implemented in GraalVM:
186-
// call:
187-
// https://github.com/awaitility/awaitility/blob/fbe16add874b4260dd240108304d5c0be84eabc8/awaitility/src/main/java/org/awaitility/core/ConditionAwaiter.java#L157
188-
// see https://github.com/oracle/graal/issues/6101 (spring boot graal native image)
189-
if (t.getClass().getName().equals("com.oracle.svm.core.jdk.UnsupportedFeatureError")
190-
|| t instanceof ConditionTimeoutException) {
191-
// Don't throw this failure since the stack is the awaitility thread, causing confusion.
192-
// Instead, just assert one more time on the test thread, which will fail with a better
193-
// stack trace.
194-
// TODO(anuraaga): There is probably a better way to do this.
195-
doAssertTraces(traceComparator, AutoConfiguredDataCapture::getSpans, assertionsList);
196-
} else {
197-
throw t;
198-
}
199-
}
200-
}
201-
202-
private static void doAssertTraces(
203-
@Nullable Comparator<List<SpanData>> traceComparator,
204-
Supplier<List<SpanData>> supplier,
205-
List<Consumer<TraceAssert>> assertionsList) {
206-
try {
207-
List<List<SpanData>> traces = waitForTraces(supplier, assertionsList.size());
208-
TelemetryDataUtil.assertScopeVersion(traces);
209-
if (traceComparator != null) {
210-
traces.sort(traceComparator);
211-
}
212-
TracesAssert.assertThat(traces).hasTracesSatisfyingExactly(assertionsList);
213-
} catch (InterruptedException | TimeoutException e) {
214-
throw new AssertionError("Error waiting for " + assertionsList.size() + " traces", e);
215-
}
216-
}
217158
}

0 commit comments

Comments
 (0)