fix scope must be closed error

Author: Cirilla-zmh

messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/AliyunMnsSdkTest.java

@@ -6,9 +6,11 @@
 package io.opentelemetry.contrib.messaging.wrappers.mns;
 
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ENVELOPE_SIZE;
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -32,6 +34,7 @@
 import io.opentelemetry.contrib.messaging.wrappers.testing.AbstractBaseTest;
 import java.nio.charset.StandardCharsets;
 import org.apache.commons.codec.binary.Base64;
+import org.assertj.core.api.AbstractAssert;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
@@ -151,7 +154,8 @@ public void assertTraces() {
                                 MESSAGING_MESSAGE_ENVELOPE_SIZE,
                                 Base64.encodeBase64(MESSAGE_BODY.getBytes(StandardCharsets.UTF_8))
                                     .length),
-                            equalTo(MESSAGING_OPERATION, "process")),
+                            equalTo(MESSAGING_OPERATION, "process"),
+                            satisfies(MESSAGING_MESSAGE_ID, AbstractAssert::isNotNull)),
                 span ->
                     span.hasName("process child")
                         .hasKind(SpanKind.INTERNAL)

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

@@ -46,32 +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();
+    Context context = span.storeInContext(Context.current());
 
-    try {
+    try (Scope scope = context.makeCurrent()) {
       runnable.run();
     } catch (Throwable t) {
-      handleEnd(span, scope, request, t);
+      handleEnd(span, context, request, t);
       throw t;
     }
 
-    handleEnd(span, scope, request, null);
+    handleEnd(span, context, 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();
+    Context context = span.storeInContext(Context.current());
 
     R result = null;
-    try {
+    try (Scope scope = context.makeCurrent()) {
       result = supplier.get();
     } catch (Throwable t) {
-      handleEnd(span, scope, request, t);
+      handleEnd(span, context, request, t);
       throw t;
     }
 
-    handleEnd(span, scope, request, null);
+    handleEnd(span, context, request, null);
     return result;
   }
 
@@ -88,14 +88,13 @@ protected Span handleStart(REQUEST request) {
     return spanBuilder.setAllAttributes(builder.build()).startSpan();
   }
 
-  protected void handleEnd(Span span, Scope scope, REQUEST request, @Nullable Throwable t) {
+  protected void handleEnd(Span span, Context context, REQUEST request, @Nullable Throwable t) {
     AttributesBuilder builder = Attributes.builder();
     for (AttributesExtractor<REQUEST, Void> extractor : this.attributesExtractors) {
-      extractor.onEnd(builder, Context.current(), request, null, t);
+      extractor.onEnd(builder, context, request, null, t);
     }
-    span.setAllAttributes(builder.build());
 
-    scope.close();
+    span.setAllAttributes(builder.build());
     span.end();
   }
 

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

@@ -10,8 +10,10 @@
 import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_BODY;
 import static io.opentelemetry.contrib.messaging.wrappers.TestConstants.MESSAGE_ID;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 
@@ -37,6 +39,8 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashMap;
+
+import org.assertj.core.api.AbstractAssert;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
@@ -127,7 +131,8 @@ public void assertTraces() {
                             //  semconv library right now. It should be replaced after semconv
                             // release.
                             equalTo(AttributeKey.stringKey("messaging.client_id"), CLIENT_ID),
-                            equalTo(MESSAGING_OPERATION, "process")),
+                            equalTo(MESSAGING_OPERATION, "process"),
+                            satisfies(MESSAGING_MESSAGE_ID, AbstractAssert::isNotNull)),
                 span ->
                     span.hasName("process child")
                         .hasKind(SpanKind.INTERNAL)

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

@@ -13,6 +13,7 @@
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_KAFKA_CONSUMER_GROUP;
 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_MESSAGE_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 
@@ -151,7 +152,8 @@ 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_OPERATION, "process"),
+                            satisfies(MESSAGING_MESSAGE_ID, AbstractAssert::isNotNull)),
                 span ->
                     span.hasName("process child")
                         .hasKind(SpanKind.INTERNAL)