add test for kafka clients

Author: Cirilla-zmh

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

@@ -1,5 +1,6 @@
 package io.opentelemetry.contrib.messaging.wrappers;
 
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.context.propagation.TextMapGetter;
@@ -9,23 +10,28 @@
 import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 
+import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 
 public class DefaultMessagingProcessWrapperBuilder<REQUEST extends MessagingProcessRequest> {
 
+  @Nullable
   private OpenTelemetry openTelemetry;
 
+  @Nullable
   protected TextMapGetter<REQUEST> textMapGetter;
 
-  protected List<AttributesExtractor<REQUEST, Void>> attributesExtractors;
-
+  @CanIgnoreReturnValue
   public DefaultMessagingProcessWrapperBuilder<REQUEST> openTelemetry(OpenTelemetry openTelemetry) {
     this.openTelemetry = openTelemetry;
     return this;
   }
 
+  protected List<AttributesExtractor<REQUEST, Void>> attributesExtractors;
+
+  @CanIgnoreReturnValue
   public DefaultMessagingProcessWrapperBuilder<REQUEST> textMapGetter(TextMapGetter<REQUEST> textMapGetter) {
     this.textMapGetter = textMapGetter;
     return this;
@@ -35,22 +41,26 @@ public DefaultMessagingProcessWrapperBuilder<REQUEST> textMapGetter(TextMapGette
    * This method overrides the original items.
    * <p>See {@link DefaultMessagingProcessWrapperBuilder#addAttributesExtractor} if you just want to append one.</p>
    * */
+  @CanIgnoreReturnValue
   public DefaultMessagingProcessWrapperBuilder<REQUEST> attributesExtractors(
       Collection<AttributesExtractor<REQUEST, Void>> attributesExtractors) {
     this.attributesExtractors = new ArrayList<>();
     this.attributesExtractors.addAll(attributesExtractors);
     return this;
   }
 
+  @CanIgnoreReturnValue
   public DefaultMessagingProcessWrapperBuilder<REQUEST> addAttributesExtractor(
       AttributesExtractor<REQUEST, Void> attributesExtractor) {
     this.attributesExtractors.add(attributesExtractor);
     return this;
   }
 
   public MessagingProcessWrapper<REQUEST> build() {
-    return new MessagingProcessWrapper<>(this.openTelemetry == null ? GlobalOpenTelemetry.get() : this.openTelemetry,
-            this.textMapGetter, this.attributesExtractors);
+    return new MessagingProcessWrapper<>(
+        this.openTelemetry == null ? GlobalOpenTelemetry.get() : this.openTelemetry,
+        this.textMapGetter == null ? NoopTextMapGetter.create() : this.textMapGetter,
+        this.attributesExtractors);
   }
 
   protected DefaultMessagingProcessWrapperBuilder() {

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

@@ -77,23 +77,23 @@ protected Span handleStart(REQUEST request) {
     return spanBuilder.setAllAttributes(builder.build()).startSpan();
   }
 
-  protected void handleEnd(Span span, REQUEST request, Throwable t) {
+  protected void handleEnd(Span span, 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.end();
   }
 
-  protected String getDefaultSpanName(String destination) {
+  protected String getDefaultSpanName(@Nullable String destination) {
     if (destination == null) {
       destination = "unknown";
     }
     return OPERATION_NAME + " " + destination;
   }
 
   protected MessagingProcessWrapper(OpenTelemetry openTelemetry,
-                          @Nullable TextMapGetter<REQUEST> textMapGetter,
+                          TextMapGetter<REQUEST> textMapGetter,
                           List<AttributesExtractor<REQUEST, Void>> attributesExtractors) {
     this.textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
     this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE + "-" + INSTRUMENTATION_VERSION);

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

@@ -0,0 +1,26 @@
+package io.opentelemetry.contrib.messaging.wrappers;
+
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest;
+import javax.annotation.Nullable;
+import java.util.Collections;
+
+public class NoopTextMapGetter<REQUEST extends MessagingProcessRequest> implements TextMapGetter<REQUEST> {
+
+  public static <REQUEST extends MessagingProcessRequest> TextMapGetter<REQUEST> create() {
+    return new NoopTextMapGetter<>();
+  }
+
+  @Override
+  public Iterable<String> keys(REQUEST request) {
+    return Collections.emptyList();
+  }
+
+  @Nullable
+  @Override
+  public String get(@Nullable REQUEST request, String s) {
+    return null;
+  }
+
+  private NoopTextMapGetter() {}
+}

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

@@ -1,5 +1,6 @@
 package io.opentelemetry.contrib.messaging.wrappers.semconv;
 
+import javax.annotation.Nullable;
 import java.util.List;
 
 import static java.util.Collections.emptyList;
@@ -13,29 +14,39 @@ public interface MessagingProcessRequest {
 
   String getSystem();
 
+  @Nullable
   String getDestination();
 
+  @Nullable
   String getDestinationTemplate();
 
   boolean isTemporaryDestination();
 
   boolean isAnonymousDestination();
 
+  @Nullable
   String getConversationId();
 
+  @Nullable
   Long getMessageBodySize();
 
+  @Nullable
   Long getMessageEnvelopeSize();
 
+  @Nullable
   String getMessageId();
 
+  @Nullable
   default String getClientId() {
     return null;
   }
+
+  @Nullable
   default Long getBatchMessageCount() {
     return null;
   }
 
+  @Nullable
   default String getDestinationPartitionId() {
     return null;
   }

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

@@ -16,10 +16,22 @@ dependencies {
 
   compileOnly("org.apache.kafka:kafka-clients:0.11.0.0")
 
+  testImplementation("org.apache.kafka:kafka-clients:0.11.0.0")
+  testImplementation("io.opentelemetry.instrumentation:opentelemetry-kafka-clients-2.6")
+
+  testAnnotationProcessor("com.google.auto.service:auto-service")
+  testCompileOnly("com.google.auto.service:auto-service-annotations")
+  testImplementation("org.junit.jupiter:junit-jupiter-api")
+  testImplementation("org.junit.jupiter:junit-jupiter-params")
+  testImplementation("org.testcontainers:kafka")
+  testImplementation("org.testcontainers:junit-jupiter")
+
   testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
-  testImplementation("io.opentelemetry:opentelemetry-sdk-trace")
+  testImplementation("io.opentelemetry:opentelemetry-sdk")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
-
   testImplementation("io.opentelemetry:opentelemetry-sdk-extension-incubator")
-  testImplementation("uk.org.webcompere:system-stubs-jupiter:2.0.3")
+  testImplementation("io.opentelemetry.semconv:opentelemetry-semconv")
+  testImplementation("io.opentelemetry.semconv:opentelemetry-semconv-incubating")
+  testImplementation("io.opentelemetry:opentelemetry-exporter-logging")
+  testImplementation("io.opentelemetry.javaagent:opentelemetry-testing-common")
 }

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

@@ -5,7 +5,6 @@
 
 import javax.annotation.Nullable;
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -14,15 +13,18 @@ public class KafkaProcessRequest implements MessagingProcessRequest {
 
   private final ConsumerRecord<?, ?> consumerRecord;
 
+  @Nullable
   private final String clientId;
 
+  @Nullable
   private final String consumerGroup;
 
   public static KafkaProcessRequest of(ConsumerRecord<?, ?> consumerRecord) {
     return of(consumerRecord, null, null);
   }
 
-  public static KafkaProcessRequest of(ConsumerRecord<?, ?> consumerRecord, String consumerGroup, String clientId) {
+  public static KafkaProcessRequest of(ConsumerRecord<?, ?> consumerRecord,
+      @Nullable String consumerGroup, @Nullable String clientId) {
     return new KafkaProcessRequest(consumerRecord, consumerGroup, clientId);
   }
 
@@ -31,11 +33,9 @@ public String getSystem() {
     return "kafka";
   }
 
+  @Nullable
   @Override
   public String getDestination() {
-    if (this.consumerRecord == null) {
-      return null;
-    }
     return this.consumerRecord.topic();
   }
 
@@ -64,9 +64,6 @@ public String getConversationId() {
   @Nullable
   @Override
   public Long getMessageBodySize() {
-    if (this.consumerRecord == null) {
-      return null;
-    }
     long size = this.consumerRecord.serializedValueSize();
     return size >= 0 ? size : null;
   }
@@ -97,9 +94,6 @@ public Long getBatchMessageCount() {
 
   @Override
   public List<String> getMessageHeader(String name) {
-    if (this.consumerRecord == null) {
-      return Collections.emptyList();
-    }
     return StreamSupport.stream(this.consumerRecord.headers().headers(name).spliterator(), false)
         .map(header -> new String(header.value(), StandardCharsets.UTF_8))
         .collect(Collectors.toList());
@@ -114,7 +108,8 @@ public String getConsumerGroup() {
     return consumerRecord;
   }
 
-  private KafkaProcessRequest(ConsumerRecord<?, ?> consumerRecord, String consumerGroup, String clientId) {
+  private KafkaProcessRequest(ConsumerRecord<?, ?> consumerRecord,
+      @Nullable String consumerGroup, @Nullable String clientId) {
     this.consumerRecord = consumerRecord;
     this.consumerGroup = consumerGroup;
     this.clientId = clientId;