Skip to content

Commit b42db0a

Browse files
jasondleejasondlee
committed
Update for MP Telemetry 2.1
Bump version numbers Fix semantic convention issues
1 parent 2979fa5 commit b42db0a

File tree

10 files changed

+92
-50
lines changed

10 files changed

+92
-50
lines changed

‎pom.xml‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -89,8 +89,9 @@
8989

9090
<kafka.version>4.0.0</kafka.version>
9191

92-
<opentelemetry.instrumentation.version>2.5.0-alpha</opentelemetry.instrumentation.version>
93-
<opentelemetry-semconv.version>1.25.0-alpha</opentelemetry-semconv.version>
92+
<opentelemetry.instrumentation.version>2.14.0</opentelemetry.instrumentation.version>
93+
<opentelemetry.version>1.48.0</opentelemetry.version>
94+
<opentelemetry-semconv.version>1.32.0</opentelemetry-semconv.version>
9495

9596
<smallrye-vertx-mutiny-clients.version>3.19.1</smallrye-vertx-mutiny-clients.version>
9697
<smallrye-reactive-converters.version>3.0.0</smallrye-reactive-converters.version>
@@ -286,10 +287,17 @@
286287
</dependency>
287288

288289
<!-- This BOM includes the opentelemetry-bom and the opentelemetry-bom-alpha -->
290+
<dependency>
291+
<groupId>io.opentelemetry</groupId>
292+
<artifactId>opentelemetry-bom</artifactId>
293+
<version>${opentelemetry.version}</version>
294+
<type>pom</type>
295+
<scope>import</scope>
296+
</dependency>
289297
<dependency>
290298
<groupId>io.opentelemetry.instrumentation</groupId>
291299
<artifactId>opentelemetry-instrumentation-bom-alpha</artifactId>
292-
<version>${opentelemetry.instrumentation.version}</version>
300+
<version>${opentelemetry.instrumentation.version}-alpha</version>
293301
<type>pom</type>
294302
<scope>import</scope>
295303
</dependency>
@@ -368,6 +376,11 @@
368376
</exclusion>
369377
</exclusions>
370378
</dependency>
379+
<dependency>
380+
<groupId>io.opentelemetry.semconv</groupId>
381+
<artifactId>opentelemetry-semconv-incubating</artifactId>
382+
<version>${opentelemetry-semconv.version}-alpha</version>
383+
</dependency>
371384
</dependencies>
372385
</dependencyManagement>
373386

‎smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/TracingAmqpToAppToAmqpTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,9 @@
22

33
import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
44
import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
5-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_DESTINATION_NAME;
6-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_OPERATION;
7-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
5+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
6+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
7+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
88
import static java.util.stream.Collectors.toList;
99
import static org.assertj.core.api.Assertions.assertThat;
1010
import static org.awaitility.Awaitility.await;

‎smallrye-reactive-messaging-jms/src/test/java/io/smallrye/reactive/messaging/jms/tracing/TracingPropagationTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
package io.smallrye.reactive.messaging.jms.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_DESTINATION_NAME;
4-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_OPERATION;
5-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
3+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
4+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
5+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
66
import static java.util.stream.Collectors.toList;
77
import static org.assertj.core.api.Assertions.assertThat;
88
import static org.awaitility.Awaitility.await;

‎smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/tracing/KafkaAttributesExtractor.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
package io.smallrye.reactive.messaging.kafka.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CONSUMER_ID;
4-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_CONSUMER_GROUP;
5-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET;
6-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_PARTITION;
3+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_CLIENT_ID;
4+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_CONSUMER_GROUP_NAME;
5+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID;
6+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_OFFSET;
77

88
import java.util.Collections;
99
import java.util.List;
@@ -14,6 +14,7 @@
1414
import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
1515

1616
public class KafkaAttributesExtractor implements AttributesExtractor<KafkaTrace, Void> {
17+
1718
private final MessagingAttributesGetter<KafkaTrace, Void> messagingAttributesGetter;
1819

1920
public KafkaAttributesExtractor() {
@@ -23,10 +24,10 @@ public KafkaAttributesExtractor() {
2324
@Override
2425
public void onStart(final AttributesBuilder attributes, final Context parentContext, final KafkaTrace kafkaTrace) {
2526
if (kafkaTrace.getPartition() != -1) {
26-
attributes.put(MESSAGING_KAFKA_PARTITION, kafkaTrace.getPartition());
27+
attributes.put(MESSAGING_DESTINATION_PARTITION_ID, Integer.toString(kafkaTrace.getPartition()));
2728
}
2829
if (kafkaTrace.getOffset() != -1) {
29-
attributes.put(MESSAGING_KAFKA_MESSAGE_OFFSET, kafkaTrace.getOffset());
30+
attributes.put(MESSAGING_KAFKA_OFFSET, kafkaTrace.getOffset());
3031
}
3132

3233
String groupId = kafkaTrace.getGroupId();
@@ -36,10 +37,10 @@ public void onStart(final AttributesBuilder attributes, final Context parentCont
3637
if (!clientId.isEmpty()) {
3738
consumerId += " - " + clientId;
3839
}
39-
attributes.put(MESSAGING_CONSUMER_ID, consumerId);
40+
attributes.put(MESSAGING_CLIENT_ID, consumerId);
4041
}
4142
if (groupId != null) {
42-
attributes.put(MESSAGING_KAFKA_CONSUMER_GROUP, groupId);
43+
attributes.put(MESSAGING_CONSUMER_GROUP_NAME, groupId);
4344
}
4445
}
4546

‎smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/tracing/TracingPropagationTest.java‎

Lines changed: 43 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,24 @@
11
package io.smallrye.reactive.messaging.kafka.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CLIENT_ID;
4-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_DESTINATION_NAME;
5-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET;
6-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_OPERATION;
7-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
3+
import static io.opentelemetry.api.common.AttributeKey.stringKey;
4+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
5+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_OFFSET;
6+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
7+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
88
import static io.smallrye.reactive.messaging.kafka.companion.RecordQualifiers.until;
99
import static java.util.stream.Collectors.toList;
1010
import static org.assertj.core.api.Assertions.assertThat;
1111
import static org.awaitility.Awaitility.await;
1212
import static org.junit.jupiter.api.Assertions.assertEquals;
13+
import static org.junit.jupiter.api.Assertions.assertTrue;
1314

1415
import java.net.URI;
1516
import java.nio.charset.StandardCharsets;
1617
import java.time.Duration;
1718
import java.util.ArrayList;
19+
import java.util.Arrays;
1820
import java.util.List;
21+
import java.util.Objects;
1922
import java.util.concurrent.CompletionStage;
2023
import java.util.concurrent.CopyOnWriteArrayList;
2124
import java.util.concurrent.Flow;
@@ -37,6 +40,8 @@
3740
import org.junit.jupiter.api.Test;
3841

3942
import io.opentelemetry.api.GlobalOpenTelemetry;
43+
import io.opentelemetry.api.common.AttributeKey;
44+
import io.opentelemetry.api.common.Attributes;
4045
import io.opentelemetry.api.trace.Span;
4146
import io.opentelemetry.api.trace.SpanId;
4247
import io.opentelemetry.api.trace.SpanKind;
@@ -53,6 +58,7 @@
5358
import io.opentelemetry.sdk.trace.data.SpanData;
5459
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
5560
import io.opentelemetry.sdk.trace.samplers.Sampler;
61+
import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes;
5662
import io.smallrye.mutiny.Multi;
5763
import io.smallrye.reactive.messaging.ce.OutgoingCloudEventMetadata;
5864
import io.smallrye.reactive.messaging.kafka.base.KafkaCompanionTestBase;
@@ -117,8 +123,9 @@ public void testFromAppToKafka() {
117123
assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
118124
assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
119125
assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
120-
assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
121-
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
126+
checkAttribute(span.getAttributes(), "kafka-producer-kafka",
127+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
128+
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_OFFSET));
122129

123130
assertEquals(topic + " publish", span.getName());
124131
});
@@ -155,9 +162,9 @@ public void testFromAppToKafkaWithStructuredCloudEvents() {
155162
assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
156163
assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
157164
assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
158-
assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
159-
assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
160-
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
165+
checkAttribute(span.getAttributes(), "kafka-producer-kafka",
166+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
167+
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_OFFSET));
161168
assertEquals(topic + " publish", span.getName());
162169
});
163170
}
@@ -193,9 +200,9 @@ public void testFromAppToKafkaWithBinaryCloudEvents() {
193200
assertEquals("kafka", span.getAttributes().get(MESSAGING_SYSTEM));
194201
assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
195202
assertEquals(topic, span.getAttributes().get(MESSAGING_DESTINATION_NAME));
196-
assertEquals("publish", span.getAttributes().get(MESSAGING_OPERATION));
197-
assertEquals("kafka-producer-kafka", span.getAttributes().get(MESSAGING_CLIENT_ID));
198-
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
203+
checkAttribute(span.getAttributes(), "kafka-producer-kafka",
204+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
205+
assertEquals(0, span.getAttributes().get(MESSAGING_KAFKA_OFFSET));
199206
assertEquals(topic + " publish", span.getName());
200207
});
201208
}
@@ -240,8 +247,9 @@ public void testFromKafkaToAppToKafka() {
240247
assertEquals("kafka", consumer.getAttributes().get(MESSAGING_SYSTEM));
241248
assertEquals("receive", consumer.getAttributes().get(MESSAGING_OPERATION));
242249
assertEquals(parentTopic, consumer.getAttributes().get(MESSAGING_DESTINATION_NAME));
243-
assertEquals("kafka-consumer-source", consumer.getAttributes().get(MESSAGING_CLIENT_ID));
244-
assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
250+
checkAttribute(consumer.getAttributes(), "kafka-consumer-source",
251+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
252+
assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_OFFSET));
245253
assertEquals(parentTopic + " receive", consumer.getName());
246254

247255
SpanData producer = spans.stream().filter(spanData -> spanData.getParentSpanId().equals(consumer.getSpanId()))
@@ -252,8 +260,9 @@ public void testFromKafkaToAppToKafka() {
252260
assertEquals("publish", producer.getAttributes().get(MESSAGING_OPERATION));
253261
assertEquals(resultTopic, producer.getAttributes().get(MESSAGING_DESTINATION_NAME));
254262
assertEquals("publish", producer.getAttributes().get(MESSAGING_OPERATION));
255-
assertEquals("kafka-producer-kafka", producer.getAttributes().get(MESSAGING_CLIENT_ID));
256-
assertEquals(0, producer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
263+
checkAttribute(producer.getAttributes(), "kafka-producer-kafka",
264+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
265+
assertEquals(0, producer.getAttributes().get(MESSAGING_KAFKA_OFFSET));
257266
assertEquals(resultTopic + " publish", producer.getName());
258267
});
259268
}
@@ -306,12 +315,27 @@ public void testFromKafkaToAppWithParentSpan() {
306315
assertEquals("kafka", consumer.getAttributes().get(MESSAGING_SYSTEM));
307316
assertEquals("receive", consumer.getAttributes().get(MESSAGING_OPERATION));
308317
assertEquals(parentTopic, consumer.getAttributes().get(MESSAGING_DESTINATION_NAME));
309-
assertEquals("kafka-consumer-stuff", consumer.getAttributes().get(MESSAGING_CLIENT_ID));
310-
assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_MESSAGE_OFFSET));
318+
checkAttribute(consumer.getAttributes(), "kafka-consumer-stuff",
319+
stringKey("messaging.client_id"), MessagingIncubatingAttributes.MESSAGING_CLIENT_ID);
320+
assertEquals(0, consumer.getAttributes().get(MESSAGING_KAFKA_OFFSET));
311321
assertEquals(parentTopic + " receive", consumer.getName());
312322
});
313323
}
314324

325+
/**
326+
* Currently, Kafka creates spans using an older version of OpenTelemetry's semantic conventions. This method
327+
* allows us to check old and new keys. It loops through the keys, making sure at least one returns the expected
328+
* value.
329+
*
330+
* @param attributes The span attributes to check
331+
* @param expected The expected value to be found
332+
* @param keys The keys under which <code>expected</code> might be found
333+
*/
334+
private void checkAttribute(Attributes attributes, String expected, AttributeKey<?>... keys) {
335+
assertTrue(
336+
Arrays.stream(keys).anyMatch(key -> Objects.equals(expected, attributes.get(key))));
337+
}
338+
315339
@Test
316340
public void testFromKafkaToAppWithNoParent() {
317341
MyAppReceivingData bean = runApplication(

‎smallrye-reactive-messaging-otel/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,10 @@
2929
<groupId>io.opentelemetry.semconv</groupId>
3030
<artifactId>opentelemetry-semconv</artifactId>
3131
</dependency>
32+
<dependency>
33+
<groupId>io.opentelemetry.semconv</groupId>
34+
<artifactId>opentelemetry-semconv-incubating</artifactId>
35+
</dependency>
3236
<dependency>
3337
<groupId>io.opentelemetry.instrumentation</groupId>
3438
<artifactId>opentelemetry-instrumentation-api</artifactId>

‎smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/tracing/PulsarAttributesExtractor.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package io.smallrye.reactive.messaging.pulsar.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_CONSUMER_ID;
3+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_CLIENT_ID;
44

55
import java.util.Collections;
66
import java.util.List;
@@ -20,7 +20,7 @@ public PulsarAttributesExtractor() {
2020
@Override
2121
public void onStart(final AttributesBuilder attributes, final Context parentContext, final PulsarTrace pulsarTrace) {
2222
String consumerName = pulsarTrace.getConsumerName();
23-
attributes.put(MESSAGING_CONSUMER_ID, consumerName);
23+
attributes.put(MESSAGING_CLIENT_ID, consumerName);
2424
}
2525

2626
@Override

‎smallrye-reactive-messaging-pulsar/src/test/java/io/smallrye/reactive/messaging/pulsar/tracing/TracingPropagationTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package io.smallrye.reactive.messaging.pulsar.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
3+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
44
import static java.util.stream.Collectors.toList;
55
import static org.assertj.core.api.Assertions.assertThat;
66
import static org.awaitility.Awaitility.await;

‎smallrye-reactive-messaging-rabbitmq/src/main/java/io/smallrye/reactive/messaging/rabbitmq/tracing/RabbitMQTraceAttributesExtractor.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package io.smallrye.reactive.messaging.rabbitmq.tracing;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_RABBITMQ_ROUTING_KEY;
3+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY;
44

55
import java.util.Collections;
66
import java.util.List;
@@ -25,7 +25,7 @@ public MessagingAttributesGetter<RabbitMQTrace, Void> getMessagingAttributesGett
2525
public void onStart(
2626
final AttributesBuilder attributes,
2727
final Context parentContext, final RabbitMQTrace rabbitMQTrace) {
28-
attributes.put(MESSAGING_RABBITMQ_ROUTING_KEY, rabbitMQTrace.getRoutingKey());
28+
attributes.put(MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY, rabbitMQTrace.getRoutingKey());
2929
}
3030

3131
@Override

‎smallrye-reactive-messaging-rabbitmq/src/test/java/io/smallrye/reactive/messaging/rabbitmq/TracingTest.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
11
package io.smallrye.reactive.messaging.rabbitmq;
22

3-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_DESTINATION_NAME;
4-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_OPERATION;
5-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_PROTOCOL;
6-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_PROTOCOL_VERSION;
7-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_RABBITMQ_ROUTING_KEY;
8-
import static io.opentelemetry.semconv.SemanticAttributes.MESSAGING_SYSTEM;
3+
import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PROTOCOL_NAME;
4+
import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PROTOCOL_VERSION;
5+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
6+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
7+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY;
8+
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
99
import static java.util.concurrent.TimeUnit.SECONDS;
1010
import static java.util.stream.Collectors.toList;
1111
import static java.util.stream.Collectors.toSet;
@@ -103,10 +103,10 @@ void incoming() {
103103
assertEquals(SpanKind.CONSUMER, consumer.getKind());
104104
assertEquals("rabbitmq", consumer.getAttributes().get(MESSAGING_SYSTEM));
105105
assertEquals("receive", consumer.getAttributes().get(MESSAGING_OPERATION));
106-
assertEquals("normal", consumer.getAttributes().get(MESSAGING_RABBITMQ_ROUTING_KEY));
106+
assertEquals("normal", consumer.getAttributes().get(MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY));
107107
assertEquals(queueName, consumer.getAttributes().get(MESSAGING_DESTINATION_NAME));
108-
assertNull(consumer.getAttributes().get(MESSAGING_PROTOCOL));
109-
assertNull(consumer.getAttributes().get(MESSAGING_PROTOCOL_VERSION));
108+
assertNull(consumer.getAttributes().get(NETWORK_PROTOCOL_NAME));
109+
assertNull(consumer.getAttributes().get(NETWORK_PROTOCOL_VERSION));
110110
assertEquals(queueName + " receive", consumer.getName());
111111
});
112112
}

0 commit comments

Comments
 (0)