Skip to content

Commit 8b235e0

Browse files
trasktrask
committed
Fix Kafka messaging semconv test expectations
1 parent adccb3b commit 8b235e0

2 files changed

Lines changed: 71 additions & 19 deletions

File tree

  • instrumentation/kafka/kafka-clients/kafka-clients-0.11

‎instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/v0_11/KafkaClientDefaultTest.java‎

Lines changed: 49 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
package io.opentelemetry.javaagent.instrumentation.kafkaclients.v0_11;
77

8+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
89
import static io.opentelemetry.instrumentation.testing.util.TelemetryDataUtil.orderByRootSpanKind;
910
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
1011
import static java.nio.charset.StandardCharsets.UTF_8;
@@ -82,7 +83,10 @@ void testKafkaProducerAndConsumerSpan(boolean testHeaders) throws Exception {
8283
trace.hasSpansSatisfyingExactly(
8384
span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
8485
span ->
85-
span.hasName(SHARED_TOPIC + " publish")
86+
span.hasName(
87+
emitStableMessagingSemconv()
88+
? "publish " + SHARED_TOPIC
89+
: SHARED_TOPIC + " publish")
8690
.hasKind(SpanKind.PRODUCER)
8791
.hasParent(trace.getSpan(0))
8892
.hasAttributesSatisfyingExactly(sendAttributes("10", greeting, testHeaders)),
@@ -95,12 +99,18 @@ void testKafkaProducerAndConsumerSpan(boolean testHeaders) throws Exception {
9599
trace ->
96100
trace.hasSpansSatisfyingExactly(
97101
span ->
98-
span.hasName(SHARED_TOPIC + " receive")
102+
span.hasName(
103+
emitStableMessagingSemconv()
104+
? "receive " + SHARED_TOPIC
105+
: SHARED_TOPIC + " receive")
99106
.hasKind(SpanKind.CONSUMER)
100107
.hasNoParent()
101108
.hasAttributesSatisfyingExactly(receiveAttributes(testHeaders)),
102109
span ->
103-
span.hasName(SHARED_TOPIC + " process")
110+
span.hasName(
111+
emitStableMessagingSemconv()
112+
? "process " + SHARED_TOPIC
113+
: SHARED_TOPIC + " process")
104114
.hasKind(SpanKind.CONSUMER)
105115
.hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
106116
.hasParent(trace.getSpan(0))
@@ -129,7 +139,10 @@ void testPassThroughTombstone() throws Exception {
129139
trace -> {
130140
trace.hasSpansSatisfyingExactly(
131141
span ->
132-
span.hasName(SHARED_TOPIC + " publish")
142+
span.hasName(
143+
emitStableMessagingSemconv()
144+
? "publish " + SHARED_TOPIC
145+
: SHARED_TOPIC + " publish")
133146
.hasKind(SpanKind.PRODUCER)
134147
.hasNoParent()
135148
.hasAttributesSatisfyingExactly(sendAttributes(null, null, false)));
@@ -138,12 +151,18 @@ void testPassThroughTombstone() throws Exception {
138151
trace ->
139152
trace.hasSpansSatisfyingExactly(
140153
span ->
141-
span.hasName(SHARED_TOPIC + " receive")
154+
span.hasName(
155+
emitStableMessagingSemconv()
156+
? "receive " + SHARED_TOPIC
157+
: SHARED_TOPIC + " receive")
142158
.hasKind(SpanKind.CONSUMER)
143159
.hasNoParent()
144160
.hasAttributesSatisfyingExactly(receiveAttributes(false)),
145161
span ->
146-
span.hasName(SHARED_TOPIC + " process")
162+
span.hasName(
163+
emitStableMessagingSemconv()
164+
? "process " + SHARED_TOPIC
165+
: SHARED_TOPIC + " process")
147166
.hasKind(SpanKind.CONSUMER)
148167
.hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
149168
.hasParent(trace.getSpan(0))
@@ -188,7 +207,10 @@ void testRecordsWithTopicPartitionKafkaConsume(boolean testListIterator) throws
188207
trace -> {
189208
trace.hasSpansSatisfyingExactly(
190209
span ->
191-
span.hasName(SHARED_TOPIC + " publish")
210+
span.hasName(
211+
emitStableMessagingSemconv()
212+
? "publish " + SHARED_TOPIC
213+
: SHARED_TOPIC + " publish")
192214
.hasKind(SpanKind.PRODUCER)
193215
.hasNoParent()
194216
.hasAttributesSatisfyingExactly(sendAttributes(null, greeting, false)));
@@ -197,12 +219,18 @@ void testRecordsWithTopicPartitionKafkaConsume(boolean testListIterator) throws
197219
trace ->
198220
trace.hasSpansSatisfyingExactly(
199221
span ->
200-
span.hasName(SHARED_TOPIC + " receive")
222+
span.hasName(
223+
emitStableMessagingSemconv()
224+
? "receive " + SHARED_TOPIC
225+
: SHARED_TOPIC + " receive")
201226
.hasKind(SpanKind.CONSUMER)
202227
.hasNoParent()
203228
.hasAttributesSatisfyingExactly(receiveAttributes(false)),
204229
span ->
205-
span.hasName(SHARED_TOPIC + " process")
230+
span.hasName(
231+
emitStableMessagingSemconv()
232+
? "process " + SHARED_TOPIC
233+
: SHARED_TOPIC + " process")
206234
.hasKind(SpanKind.CONSUMER)
207235
.hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
208236
.hasParent(trace.getSpan(0))
@@ -253,7 +281,10 @@ void testKafkaHeaderNull() throws Exception {
253281
trace.hasSpansSatisfyingExactly(
254282
span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
255283
span ->
256-
span.hasName(SHARED_TOPIC + " publish")
284+
span.hasName(
285+
emitStableMessagingSemconv()
286+
? "publish " + SHARED_TOPIC
287+
: SHARED_TOPIC + " publish")
257288
.hasKind(SpanKind.PRODUCER)
258289
.hasParent(trace.getSpan(0))
259290
.hasAttributesSatisfyingExactly(sendAttributes("10", greeting, false)),
@@ -266,12 +297,18 @@ void testKafkaHeaderNull() throws Exception {
266297
trace ->
267298
trace.hasSpansSatisfyingExactly(
268299
span ->
269-
span.hasName(SHARED_TOPIC + " receive")
300+
span.hasName(
301+
emitStableMessagingSemconv()
302+
? "receive " + SHARED_TOPIC
303+
: SHARED_TOPIC + " receive")
270304
.hasKind(SpanKind.CONSUMER)
271305
.hasNoParent()
272306
.hasAttributesSatisfyingExactly(receiveAttributes(false)),
273307
span ->
274-
span.hasName(SHARED_TOPIC + " process")
308+
span.hasName(
309+
emitStableMessagingSemconv()
310+
? "process " + SHARED_TOPIC
311+
: SHARED_TOPIC + " process")
275312
.hasKind(SpanKind.CONSUMER)
276313
.hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
277314
.hasParent(trace.getSpan(0))

‎instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/KafkaClientBaseTest.java‎

Lines changed: 22 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
1515
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
1616
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
17+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_CLIENT_ID;
1718
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
1819
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID;
1920
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_CONSUMER_GROUP;
@@ -23,6 +24,8 @@
2324
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_OFFSET;
2425
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
2526
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
27+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION_NAME;
28+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION_TYPE;
2629
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
2730
import static java.nio.charset.StandardCharsets.UTF_8;
2831
import static java.util.Arrays.asList;
@@ -74,7 +77,7 @@ public abstract class KafkaClientBaseTest {
7477
@RegisterExtension final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
7578

7679
protected static final String SHARED_TOPIC = "shared.topic";
77-
protected static final AttributeKey<String> MESSAGING_CLIENT_ID =
80+
private static final AttributeKey<String> MESSAGING_CLIENT_ID_OLD =
7881
AttributeKey.stringKey("messaging.client_id");
7982

8083
private KafkaContainer kafka;
@@ -181,8 +184,12 @@ protected static List<AttributeAssertion> sendAttributes(
181184
asList(
182185
equalTo(MESSAGING_SYSTEM, "kafka"),
183186
equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
184-
equalTo(MESSAGING_OPERATION, "publish"),
185-
satisfies(MESSAGING_CLIENT_ID, val -> val.startsWith("producer")),
187+
equalTo(MESSAGING_OPERATION, emitOldMessagingSemconv() ? "publish" : null),
188+
equalTo(MESSAGING_OPERATION_NAME, emitStableMessagingSemconv() ? "publish" : null),
189+
equalTo(MESSAGING_OPERATION_TYPE, emitStableMessagingSemconv() ? "send" : null),
190+
satisfies(
191+
emitStableMessagingSemconv() ? MESSAGING_CLIENT_ID : MESSAGING_CLIENT_ID_OLD,
192+
val -> val.startsWith("producer")),
186193
satisfies(MESSAGING_DESTINATION_PARTITION_ID, AbstractStringAssert::isNotEmpty)));
187194
if (emitOldMessagingSemconv()) {
188195
assertions.add(satisfies(MESSAGING_KAFKA_MESSAGE_OFFSET, AbstractLongAssert::isNotNegative));
@@ -215,8 +222,12 @@ protected static List<AttributeAssertion> receiveAttributes(boolean testHeaders)
215222
asList(
216223
equalTo(MESSAGING_SYSTEM, "kafka"),
217224
equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
218-
equalTo(MESSAGING_OPERATION, "receive"),
219-
satisfies(MESSAGING_CLIENT_ID, val -> val.startsWith("consumer")),
225+
equalTo(MESSAGING_OPERATION, emitOldMessagingSemconv() ? "receive" : null),
226+
equalTo(MESSAGING_OPERATION_NAME, emitStableMessagingSemconv() ? "receive" : null),
227+
equalTo(MESSAGING_OPERATION_TYPE, emitStableMessagingSemconv() ? "receive" : null),
228+
satisfies(
229+
emitStableMessagingSemconv() ? MESSAGING_CLIENT_ID : MESSAGING_CLIENT_ID_OLD,
230+
val -> val.startsWith("consumer")),
220231
satisfies(MESSAGING_BATCH_MESSAGE_COUNT, AbstractLongAssert::isPositive)));
221232
// consumer group is not available in version 0.11
222233
if (testLatestDeps()) {
@@ -236,8 +247,12 @@ protected static List<AttributeAssertion> processAttributes(
236247
asList(
237248
equalTo(MESSAGING_SYSTEM, "kafka"),
238249
equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
239-
equalTo(MESSAGING_OPERATION, "process"),
240-
satisfies(MESSAGING_CLIENT_ID, val -> val.startsWith("consumer")),
250+
equalTo(MESSAGING_OPERATION, emitOldMessagingSemconv() ? "process" : null),
251+
equalTo(MESSAGING_OPERATION_NAME, emitStableMessagingSemconv() ? "process" : null),
252+
equalTo(MESSAGING_OPERATION_TYPE, emitStableMessagingSemconv() ? "process" : null),
253+
satisfies(
254+
emitStableMessagingSemconv() ? MESSAGING_CLIENT_ID : MESSAGING_CLIENT_ID_OLD,
255+
val -> val.startsWith("consumer")),
241256
satisfies(MESSAGING_DESTINATION_PARTITION_ID, AbstractStringAssert::isNotEmpty)));
242257
if (EXPERIMENTAL_ATTRIBUTES) {
243258
assertions.add(

0 commit comments

Comments
 (0)