Added missing messaging semantic attributes on the consumer spans (#856)

* Added missing messaging semantic attributes on the consumer spans

Signed-off-by: Paolo Patierno <[email protected]>

* Lower casing the consumer span operation name

Signed-off-by: Paolo Patierno <[email protected]>

---------

Signed-off-by: Paolo Patierno <[email protected]>

Author: ppatierno

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 0.28.0
 
 * Dependency updates (Kafka 3.6.1, Kubernetes configuration provider 1.1.2, Vert.x 4.5.0)
+* Fixed missing messaging semantic attributes to the Kafka consumer spans
 
 ## 0.27.0
 

src/main/java/io/strimzi/kafka/bridge/tracing/OpenTelemetryHandle.java

@@ -28,6 +28,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
 
@@ -90,8 +91,11 @@ private SpanBuilder getSpanBuilder(RoutingContext routingContext, String operati
 
     @Override
     public <K, V> void handleRecordSpan(ConsumerRecord<K, V> record) {
-        String operationName = record.topic() + " " + MessageOperation.RECEIVE;
+        String operationName = record.topic() + " " + MessageOperation.RECEIVE.name().toLowerCase(Locale.ROOT);
         SpanBuilder spanBuilder = get().spanBuilder(operationName);
+        spanBuilder.setAttribute(SemanticAttributes.MESSAGING_DESTINATION, record.topic());
+        spanBuilder.setAttribute(SemanticAttributes.MESSAGING_DESTINATION_KIND, SemanticAttributes.MessagingDestinationKindValues.TOPIC);
+        spanBuilder.setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "kafka");
         Context parentContext = propagator().extract(Context.current(), TracingUtil.toHeaders(record), MG);
         if (parentContext != null) {
             Span parentSpan = Span.fromContext(parentContext);