wip: addressing the feedback and removing unwanted dependencies.

Author: vasantteja

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectAttributesGetter.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+enum KafkaConnectAttributesGetter implements MessagingAttributesGetter<KafkaConnectTask, Void> {
+  INSTANCE;
+
+  @Override
+  public String getSystem(KafkaConnectTask request) {
+    return "kafka";
+  }
+
+  @Override
+  @Nullable
+  public String getDestination(KafkaConnectTask request) {
+    SinkRecord firstRecord = request.getFirstRecord();
+    return firstRecord != null ? firstRecord.topic() : null;
+  }
+
+  @Nullable
+  @Override
+  public String getDestinationTemplate(KafkaConnectTask request) {
+    return null;
+  }
+
+  @Override
+  public boolean isTemporaryDestination(KafkaConnectTask request) {
+    return false;
+  }
+
+  @Override
+  public boolean isAnonymousDestination(KafkaConnectTask request) {
+    return false;
+  }
+
+  @Override
+  @Nullable
+  public String getConversationId(KafkaConnectTask request) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageBodySize(KafkaConnectTask request) {
+    // SinkRecord doesn't expose serialized size information
+    // This would need to be calculated from the actual value, but that's expensive
+    // and not typically done in messaging instrumentations for batch processing
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageEnvelopeSize(KafkaConnectTask request) {
+    return null;
+  }
+
+  @Override
+  @Nullable
+  public String getMessageId(KafkaConnectTask request, @Nullable Void unused) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public String getClientId(KafkaConnectTask request) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getBatchMessageCount(KafkaConnectTask request, @Nullable Void unused) {
+    return (long) request.getRecords().size();
+  }
+
+  @Override
+  public List<String> getMessageHeader(KafkaConnectTask request, String name) {
+    SinkRecord firstRecord = request.getFirstRecord();
+    if (firstRecord == null || firstRecord.headers() == null) {
+      return Collections.emptyList();
+    }
+    
+    return StreamSupport.stream(firstRecord.headers().spliterator(), false)
+        .filter(header -> name.equals(header.key()) && header.value() != null)
+        .map(header -> convertHeaderValue(header))
+        .collect(Collectors.toList());
+  }
+
+  private static String convertHeaderValue(Header header) {
+    Object value = header.value();
+    if (value instanceof byte[]) {
+      return new String((byte[]) value, StandardCharsets.UTF_8);
+    }
+    return value.toString();
+  }
+}

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectBatchProcessSpanLinksExtractor.java

@@ -12,8 +12,12 @@
 import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor;
 import org.apache.kafka.connect.sink.SinkRecord;
 
-final class KafkaConnectBatchProcessSpanLinksExtractor
-    implements SpanLinksExtractor<KafkaConnectTask> {
+/**
+ * Extracts span links from Kafka Connect SinkRecord headers for batch processing scenarios.
+ * This ensures that when processing a batch of records that may come from different traces,
+ * we create links to all the original trace contexts rather than losing them.
+ */
+final class KafkaConnectBatchProcessSpanLinksExtractor implements SpanLinksExtractor<KafkaConnectTask> {
 
   private final SpanLinksExtractor<SinkRecord> singleRecordLinkExtractor;
 
@@ -23,10 +27,13 @@ final class KafkaConnectBatchProcessSpanLinksExtractor
   }
 
   @Override
-  public void extract(SpanLinksBuilder spanLinks, Context parentContext, KafkaConnectTask task) {
+  public void extract(
+      SpanLinksBuilder spanLinks, Context parentContext, KafkaConnectTask request) {
 
-    for (SinkRecord record : task.getRecords()) {
-      singleRecordLinkExtractor.extract(spanLinks, parentContext, record);
+    for (SinkRecord record : request.getRecords()) {
+      // Create a link to each record's original trace context
+      // Using Context.root() to avoid linking to the current span
+      singleRecordLinkExtractor.extract(spanLinks, Context.root(), record);
     }
   }
 }

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectSingletons.java

@@ -8,6 +8,9 @@
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.context.propagation.TextMapGetter;
 import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
 import org.apache.kafka.connect.sink.SinkRecord;
@@ -21,11 +24,22 @@ public final class KafkaConnectSingletons {
   private static final TextMapGetter<SinkRecord> SINK_RECORD_HEADER_GETTER =
       SinkRecordHeadersGetter.INSTANCE;
 
-  private static final Instrumenter<KafkaConnectTask, Void> INSTRUMENTER =
-      Instrumenter.<KafkaConnectTask, Void>builder(
-              GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME, KafkaConnectTask::getSpanName)
-          .addSpanLinksExtractor(new KafkaConnectBatchProcessSpanLinksExtractor(PROPAGATOR))
-          .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
+  private static final Instrumenter<KafkaConnectTask, Void> INSTRUMENTER;
+
+  static {
+    KafkaConnectBatchProcessSpanLinksExtractor spanLinksExtractor = new KafkaConnectBatchProcessSpanLinksExtractor(PROPAGATOR);
+
+    INSTRUMENTER =
+        Instrumenter.<KafkaConnectTask, Void>builder(
+                GlobalOpenTelemetry.get(),
+                INSTRUMENTATION_NAME,
+                MessagingSpanNameExtractor.create(KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS))
+            .addAttributesExtractor(
+                MessagingAttributesExtractor.builder(KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS)
+                    .build())
+            .addSpanLinksExtractor(spanLinksExtractor)
+            .buildInstrumenter(SpanKindExtractor.alwaysConsumer());
+  }
 
   public static Instrumenter<KafkaConnectTask, Void> instrumenter() {
     return INSTRUMENTER;

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectTask.java

@@ -20,7 +20,12 @@ public Collection<SinkRecord> getRecords() {
     return records;
   }
 
-  public static String getSpanName(KafkaConnectTask task) {
-    return "KafkaConnect.put";
+  /**
+   * Returns the first record in the batch, used for extracting destination information.
+   * Kafka Connect processes records in batches, but all records in a batch typically
+   * come from the same topic, so we use the first record for span naming.
+   */
+  public SinkRecord getFirstRecord() {
+    return records.isEmpty() ? null : records.iterator().next();
   }
 }

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/SinkRecordHeadersGetter.java

@@ -7,37 +7,42 @@
 
 import io.opentelemetry.context.propagation.TextMapGetter;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.stream.StreamSupport;
 import javax.annotation.Nullable;
 import org.apache.kafka.connect.header.Header;
-import org.apache.kafka.connect.header.Headers;
 import org.apache.kafka.connect.sink.SinkRecord;
 
+/**
+ * Extracts trace context from Kafka Connect SinkRecord headers for distributed tracing.
+ * This enables proper trace propagation from the original producer through Kafka Connect processing.
+ */
 enum SinkRecordHeadersGetter implements TextMapGetter<SinkRecord> {
   INSTANCE;
 
   @Override
   public Iterable<String> keys(SinkRecord record) {
-    Headers headers = record.headers();
-    List<String> keys = new ArrayList<>();
-    for (Header header : headers) {
-      keys.add(header.key());
+    if (record.headers() == null) {
+      return java.util.Collections.emptyList();
     }
-    return keys;
+    
+    return StreamSupport.stream(record.headers().spliterator(), false)
+        .map(Header::key)
+        .collect(java.util.stream.Collectors.toList());
   }
 
-  @Nullable
   @Override
+  @Nullable
   public String get(@Nullable SinkRecord record, String key) {
-    if (record == null) {
+    if (record == null || record.headers() == null) {
       return null;
     }
-    Headers headers = record.headers();
-    Header header = headers.lastWithName(key);
-    if (header == null) {
+
+    Header header = record.headers().lastWithName(key);
+    if (header == null || header.value() == null) {
       return null;
     }
+
+    // Convert header value to string
     Object value = header.value();
     if (value instanceof byte[]) {
       return new String((byte[]) value, StandardCharsets.UTF_8);

instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/SinkTaskInstrumentation.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
 
+import static io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6.KafkaConnectSingletons.instrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -58,11 +59,11 @@ public static void onEnter(
       }
 
       task = new KafkaConnectTask(records);
-      if (!KafkaConnectSingletons.instrumenter().shouldStart(parentContext, task)) {
+      if (!instrumenter().shouldStart(parentContext, task)) {
         return;
       }
 
-      context = KafkaConnectSingletons.instrumenter().start(parentContext, task);
+      context = instrumenter().start(parentContext, task);
       scope = context.makeCurrent();
     }
 
@@ -77,7 +78,7 @@ public static void onExit(
         return;
       }
       scope.close();
-      KafkaConnectSingletons.instrumenter().end(context, task, null, throwable);
+      instrumenter().end(context, task, null, throwable);
     }
   }
 }

instrumentation/kafka/kafka-connect-2.6/testing/build.gradle.kts

@@ -6,36 +6,23 @@ plugins {
 dependencies {
   api(project(":testing-common"))
 
-  implementation("org.apache.kafka:kafka-clients:0.11.0.0")
+  testImplementation("org.apache.kafka:kafka-clients:3.6.1")
 
-  implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-common-0.11:library"))
-  implementation(project(":instrumentation:kafka:kafka-connect-2.6:javaagent"))
-  implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-0.11:javaagent"))
-
-  testImplementation("org.apache.kafka:connect-runtime:3.6.1")
-  testImplementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-0.11:testing"))
-
-  implementation("org.testcontainers:postgresql:1.21.3") // For PostgreSQLContainer
+  testImplementation("org.testcontainers:postgresql:1.21.3") // For PostgreSQLContainer
   testImplementation("org.postgresql:postgresql:42.7.2") // PostgreSQL JDBC driver
-  implementation("org.testcontainers:mongodb:1.21.3") // For MongoDBContainer
+  testImplementation("org.testcontainers:mongodb:1.21.3") // For MongoDBContainer
   testImplementation("org.mongodb:mongodb-driver-sync:4.11.0") // MongoDB Java driver
   testImplementation("org.apache.httpcomponents:httpclient") // For HttpStatus (not httpcore)
 
   testImplementation("org.mockito:mockito-junit-jupiter:4.11.0")
 
   // Testcontainers dependencies for integration testing
-  testImplementation("io.strimzi:strimzi-test-container:0.111.0")
-  implementation("org.testcontainers:junit-jupiter")
+  testImplementation("org.testcontainers:junit-jupiter")
   testImplementation("org.awaitility:awaitility")
   testImplementation("com.squareup.okhttp3:okhttp:4.12.0")
   testImplementation("org.testcontainers:testcontainers:1.19.7")
   testImplementation("org.testcontainers:kafka:1.19.7")
-  implementation("io.rest-assured:rest-assured:5.5.5")
+  testImplementation("io.rest-assured:rest-assured:5.5.5")
   testImplementation("org.junit.jupiter:junit-jupiter:5.10.2")
-  testImplementation("org.testcontainers:junit-jupiter")
-  implementation("com.fasterxml.jackson.core:jackson-databind")
-  testImplementation("com.squareup.okhttp3:okhttp:4.12.0")
-
-  compileOnly("com.google.auto.value:auto-value-annotations")
-  annotationProcessor("com.google.auto.value:auto-value")
+  testImplementation("com.fasterxml.jackson.core:jackson-databind")
 }