improve the integration test of kafka-clients messaging wrapper

Author: Cirilla-zmh

messaging-wrappers/README.md

@@ -15,9 +15,99 @@ this tool aims to streamline the tracing and monitoring process.
 
 ## Predefined Implementations
 
-| Messaging system  | Version        | Wrapper type |
-|-------------------|----------------|--------------|
-| Aliyun mns-client | 1.3.0-SNAPSHOT | process      |
+| Messaging system  | Version Scope       | Wrapper type |
+|-------------------|---------------------|--------------|
+| kafka-clients     | `[0.11.0.0,)`       | process      |
+| aliyun mns-client | `[1.3.0-SNAPSHOT,)` | process      |
+
+## Quickstart
+
+### Step 1 Add dependencies
+
+To use OpenTelemetry in your project, you need to add the necessary dependencies. Below are the configurations for both
+Gradle and Maven.
+
+#### Gradle
+
+```kotlin
+dependencies {
+    implementation("io.opentelemetry.contrib:opentelemetry-messaging-wrappers-api")
+}
+```
+
+#### Maven
+
+```xml
+<dependency>
+    <groupId>io.opentelemetry.contrib</groupId>
+    <artifactId>opentelemetry-messaging-wrappers-api</artifactId>
+</dependency>
+```
+
+### Step 2 Initializing MessagingWrappers
+
+Below is an example of how to initialize a messaging wrapper.
+
+```java
+public class Demo {
+
+  public static MessagingProcessWrapper<MyMessagingProcessRequest> createWrapper(
+      OpenTelemetry openTelemetry,
+      MyTextMapGetter textMapGetter,
+      List<AttributesExtractor<MyMessagingProcessRequest, Void>> additionalExtractor) {
+
+    return MessagingProcessWrapper.<MyMessagingProcessRequest>defaultBuilder()
+        .openTelemetry(openTelemetry)
+        .textMapGetter(textMapGetter)
+        .addAttributesExtractors(additionalExtractor)
+        .build();
+  }
+}
+
+public class MyMessagingProcessRequest implements MessagingProcessRequest {
+  // your implementation here
+}
+
+public class MyTextMapGetter implements TextMapGetter<MyMessagingProcessRequest> {
+  // your implementation here
+}
+```
+
+For arbitrary messaging systems, you need to manually define `MessagingProcessRequest` and the corresponding `TextMapGetter`.
+You can also customize your messaging spans by adding an AttributesExtractor.
+
+For popular messaging systems, we provide pre-implemented wrappers that allow for out-of-the-box integration. We provide
+an implementation based on the OpenTelemetry semantic convention by default.
+
+```java
+public class KafkaDemo {
+  
+  public static MessagingProcessWrapper<KafkaProcessRequest> createWrapper() {
+    return KafkaHelper.processWrapperBuilder().build();
+  }
+}
+```
+
+### Step 3 Wrapping your process
+
+Once the MessagingWrapper are initialized, you can wrap your message processing logic to ensure that tracing spans are
+properly created and propagated.
+
+**P.S.** Some instrumentations may also [generate process spans](https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/docs/supported-libraries.md).
+If both are enabled, it might result in duplicate nested process spans. It is recommended to disable one of them.
+
+```java
+public class Demo {
+  
+  private static final MessagingProcessWrapper<MyMessagingProcessRequest> WRAPPER = createWrapper();
+  
+  public String consume(Message message) {
+    WRAPPER.doProcess(new MyMessagingProcessRequest(message), () -> {
+      // your processing logic
+    });
+  }
+}
+```
 
 ## Component owners
 

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

@@ -56,6 +56,13 @@ public DefaultMessagingProcessWrapperBuilder<REQUEST> addAttributesExtractor(
     return this;
   }
 
+  @CanIgnoreReturnValue
+  public DefaultMessagingProcessWrapperBuilder<REQUEST> addAttributesExtractors(
+      Collection<AttributesExtractor<REQUEST, Void>> attributesExtractor) {
+    this.attributesExtractors.addAll(attributesExtractor);
+    return this;
+  }
+
   public MessagingProcessWrapper<REQUEST> build() {
     return new MessagingProcessWrapper<>(
         this.openTelemetry == null ? GlobalOpenTelemetry.get() : this.openTelemetry,

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

@@ -16,6 +16,8 @@
 import javax.annotation.Nullable;
 import java.util.List;
 
+import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+
 public class MessagingProcessWrapper<REQUEST extends MessagingProcessRequest> {
 
   private static final String INSTRUMENTATION_SCOPE = "messaging-process-wrapper";
@@ -68,7 +70,7 @@ public <R, E extends Throwable> R doProcess(REQUEST request, ThrowingSupplier<R,
   protected Span handleStart(REQUEST request) {
     Context context = this.textMapPropagator.extract(Context.current(), request, this.textMapGetter);
     SpanBuilder spanBuilder = this.tracer.spanBuilder(getDefaultSpanName(request.getDestination()));
-    spanBuilder.setParent(context);
+    spanBuilder.setSpanKind(CONSUMER).setParent(context);
 
     AttributesBuilder builder = Attributes.builder();
     for (AttributesExtractor<REQUEST, Void> extractor : this.attributesExtractors) {
@@ -96,7 +98,7 @@ protected MessagingProcessWrapper(OpenTelemetry openTelemetry,
                           TextMapGetter<REQUEST> textMapGetter,
                           List<AttributesExtractor<REQUEST, Void>> attributesExtractors) {
     this.textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
-    this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE + "-" + INSTRUMENTATION_VERSION);
+    this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE, INSTRUMENTATION_VERSION);
     this.textMapGetter = textMapGetter;
     this.attributesExtractors = attributesExtractors;
   }

messaging-wrappers/kafka-clients/build.gradle.kts

@@ -39,7 +39,10 @@ dependencies {
 tasks {
   withType<Test>().configureEach {
     jvmArgs("-Dotel.java.global-autoconfigure.enabled=true")
-    jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
+    // TODO: According to https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#message-creation-context-as-parent-of-process-span,
+    //  process span should be the child of receive span. However, we couldn't access the trace context with receive span
+    //  in wrappers, unless we add a generic accessor for that.
+    jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=false")
     jvmArgs("-Dotel.traces.exporter=logging")
     jvmArgs("-Dotel.metrics.exporter=logging")
     jvmArgs("-Dotel.logs.exporter=logging")

messaging-wrappers/kafka-clients/src/main/java/io/opentelemetry/contrib/messaging/wrappers/kafka/semconv/KafkaConsumerAttributesExtractor.java

@@ -18,12 +18,12 @@ public final class KafkaConsumerAttributesExtractor<REQUEST extends KafkaProcess
   // copied from MessagingIncubatingAttributes
   private static final AttributeKey<String> MESSAGING_DESTINATION_PARTITION_ID =
       AttributeKey.stringKey("messaging.destination.partition.id");
-  private static final AttributeKey<String> MESSAGING_KAFKA_CONSUMER_GROUP =
-      AttributeKey.stringKey("messaging.kafka.consumer.group");
+  private static final AttributeKey<String> MESSAGING_CONSUMER_GROUP_NAME =
+      AttributeKey.stringKey("messaging.consumer.group.name");
+  private static final AttributeKey<Long> MESSAGING_KAFKA_OFFSET =
+      AttributeKey.longKey("messaging.kafka.offset");
   private static final AttributeKey<String> MESSAGING_KAFKA_MESSAGE_KEY =
       AttributeKey.stringKey("messaging.kafka.message.key");
-  private static final AttributeKey<Long> MESSAGING_KAFKA_MESSAGE_OFFSET =
-      AttributeKey.longKey("messaging.kafka.message.offset");
   private static final AttributeKey<Boolean> MESSAGING_KAFKA_MESSAGE_TOMBSTONE =
       AttributeKey.booleanKey("messaging.kafka.message.tombstone");
 
@@ -38,7 +38,7 @@ public void onStart(
     ConsumerRecord<?, ?> record = request.getRecord();
 
     attributes.put(MESSAGING_DESTINATION_PARTITION_ID, String.valueOf(record.partition()));
-    attributes.put(MESSAGING_KAFKA_MESSAGE_OFFSET, record.offset());
+    attributes.put(MESSAGING_KAFKA_OFFSET, record.offset());
 
     Object key = record.key();
     if (key != null && canSerialize(key.getClass())) {
@@ -50,7 +50,7 @@ public void onStart(
 
     String consumerGroup = request.getConsumerGroup();
     if (consumerGroup != null) {
-      attributes.put(MESSAGING_KAFKA_CONSUMER_GROUP, consumerGroup);
+      attributes.put(MESSAGING_CONSUMER_GROUP_NAME, consumerGroup);
     }
   }