fix the messaging wrappers

Author: Cirilla-zmh

messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapper.java

@@ -46,30 +46,32 @@ public static <REQUEST> MessagingProcessWrapperBuilder<REQUEST> defaultBuilder()
   public <E extends Throwable> void doProcess(REQUEST request, ThrowingRunnable<E> runnable)
       throws E {
     Span span = handleStart(request);
+    Scope scope = span.makeCurrent();
 
-    try (Scope scope = span.makeCurrent()) {
+    try {
       runnable.run();
     } catch (Throwable t) {
-      handleEnd(span, request, t);
+      handleEnd(span, scope, request, t);
       throw t;
     }
 
-    handleEnd(span, request, null);
+    handleEnd(span, scope, request, null);
   }
 
   public <R, E extends Throwable> R doProcess(REQUEST request, ThrowingSupplier<R, E> supplier)
       throws E {
     Span span = handleStart(request);
+    Scope scope = span.makeCurrent();
 
     R result = null;
-    try (Scope scope = span.makeCurrent()) {
+    try {
       result = supplier.get();
     } catch (Throwable t) {
-      handleEnd(span, request, t);
+      handleEnd(span, scope, request, t);
       throw t;
     }
 
-    handleEnd(span, request, null);
+    handleEnd(span, scope, request, null);
     return result;
   }
 
@@ -86,11 +88,14 @@ protected Span handleStart(REQUEST request) {
     return spanBuilder.setAllAttributes(builder.build()).startSpan();
   }
 
-  protected void handleEnd(Span span, REQUEST request, @Nullable Throwable t) {
+  protected void handleEnd(Span span, Scope scope, REQUEST request, @Nullable Throwable t) {
     AttributesBuilder builder = Attributes.builder();
     for (AttributesExtractor<REQUEST, Void> extractor : this.attributesExtractors) {
       extractor.onEnd(builder, Context.current(), request, null, t);
     }
+    span.setAllAttributes(builder.build());
+
+    scope.close();
     span.end();
   }
 

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

@@ -11,7 +11,6 @@
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_CONSUMER_GROUP;
-import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_MESSAGE_KEY;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
@@ -152,8 +151,7 @@ public void assertTraces() {
                                 AttributeKey.stringKey("messaging.client_id"), "test-consumer-1"),
                             satisfies(MESSAGING_KAFKA_MESSAGE_OFFSET, AbstractAssert::isNotNull),
                             equalTo(MESSAGING_KAFKA_CONSUMER_GROUP, "test"),
-                            equalTo(MESSAGING_OPERATION, "process"),
-                            equalTo(MESSAGING_KAFKA_MESSAGE_KEY, "42")),
+                            equalTo(MESSAGING_OPERATION, "process")),
                 span ->
                     span.hasName("process child")
                         .hasKind(SpanKind.INTERNAL)