open-telemetry · vasantteja · Aug 8, 2025 · Aug 15, 2025 · Aug 17, 2025 · Aug 17, 2025
@@ -595,6 +595,9 @@ targets:
     - type: gradle
       path: ./
       target: ':instrumentation:jsf:jsf-myfaces-3.0:javaagent'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:kafka:kafka-connect-2.6:javaagent'
     - type: gradle
       path: ./
       target: ':instrumentation:kafka:kafka-streams-0.11:javaagent'

@@ -0,0 +1,40 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.apache.kafka")
+    module.set("connect-api")
+    versions.set("[2.6.0,)")
+    // we use reflection to access the "pause" and "resume" methods, so we can't reference them
+    // directly, and so we can't assert that they exist at muzzle-time
+    skip("org.apache.kafka.connect.sink.SinkTaskContext")
+  }
+}
+
+dependencies {
+  implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-common-0.11:library"))
+
+  library("org.apache.kafka:connect-api:2.6.0")
+
+  testImplementation("org.apache.kafka:connect-runtime:2.6.0")
+}
+
+tasks {
+  withType<Test>().configureEach {
+    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
+
+    systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
+
+    // Enable experimental span attributes and receive telemetry for comprehensive testing
+    jvmArgs("-Dotel.instrumentation.kafka.experimental-span-attributes=true")
+    jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
+    // Set timeout for integration tests with containers
+    systemProperty("junit.jupiter.execution.timeout.default", "5m")
+  }
+
+  withType<JavaCompile>().configureEach {
+    options.compilerArgs.add("-Xlint:-deprecation")
+  }
+}
@@ -0,0 +1,32 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksBuilder;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksExtractor;
+import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+final class KafkaConnectBatchProcessSpanLinksExtractor
+    implements SpanLinksExtractor<KafkaConnectTask> {
+
+  private final SpanLinksExtractor<SinkRecord> singleRecordLinkExtractor;
+
+  KafkaConnectBatchProcessSpanLinksExtractor(TextMapPropagator propagator) {
+    this.singleRecordLinkExtractor =
+        new PropagatorBasedSpanLinksExtractor<>(propagator, SinkRecordHeadersGetter.INSTANCE);
+  }
+
+  @Override
+  public void extract(SpanLinksBuilder spanLinks, Context parentContext, KafkaConnectTask task) {
+
+    for (SinkRecord record : task.getRecords()) {
+      singleRecordLinkExtractor.extract(spanLinks, parentContext, record);
+    }
+  }
+}
@@ -0,0 +1,26 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import static java.util.Collections.singletonList;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import java.util.List;
+
+@AutoService(InstrumentationModule.class)
+public class KafkaConnectInstrumentationModule extends InstrumentationModule {
+
+  public KafkaConnectInstrumentationModule() {
+    super("kafka-connect", "kafka-connect-2.6");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return singletonList(new SinkTaskInstrumentation());
+  }
+}
@@ -0,0 +1,42 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+public final class KafkaConnectSingletons {
+
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.kafka-connect-2.6";
+  private static final TextMapPropagator PROPAGATOR =
+      GlobalOpenTelemetry.get().getPropagators().getTextMapPropagator();
+
+  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();
+
+  public static Instrumenter<KafkaConnectTask, Void> instrumenter() {
+    return INSTRUMENTER;
+  }
+
+  public static TextMapPropagator propagator() {
+    return PROPAGATOR;
+  }
+
+  public static TextMapGetter<SinkRecord> sinkRecordHeaderGetter() {
+    return SINK_RECORD_HEADER_GETTER;
+  }
+
+  private KafkaConnectSingletons() {}
+}
@@ -0,0 +1,26 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import java.util.Collection;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+public final class KafkaConnectTask {
+
+  private final Collection<SinkRecord> records;
+
+  public KafkaConnectTask(Collection<SinkRecord> records) {
+    this.records = records;
+  }
+
+  public Collection<SinkRecord> getRecords() {
+    return records;
+  }
+
+  public static String getSpanName(KafkaConnectTask task) {
+    return "KafkaConnect.put";
+  }
+}
@@ -0,0 +1,47 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import io.opentelemetry.context.propagation.TextMapGetter;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+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;
+
+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());
+    }
+    return keys;
+  }
+
+  @Nullable
+  @Override
+  public String get(@Nullable SinkRecord record, String key) {
+    if (record == null) {
+      return null;
+    }
+    Headers headers = record.headers();
+    Header header = headers.lastWithName(key);
+    if (header == null) {
+      return null;
+    }
+    Object value = header.value();
+    if (value instanceof byte[]) {
+      return new String((byte[]) value, StandardCharsets.UTF_8);
+    }
+    return value.toString();
+  }
+}
@@ -0,0 +1,83 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaconnect.v2_6;
+
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import java.util.Collection;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+public class SinkTaskInstrumentation implements TypeInstrumentation {
+
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return hasSuperType(named("org.apache.kafka.connect.sink.SinkTask"));
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        named("put").and(takesArgument(0, Collection.class)).and(isPublic()),
+        SinkTaskInstrumentation.class.getName() + "$SinkTaskPutAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class SinkTaskPutAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(
+        @Advice.Argument(0) Collection<SinkRecord> records,
+        @Advice.Local("otelTask") KafkaConnectTask task,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      Context parentContext = Java8BytecodeBridge.currentContext();
+
+      // Extract context from first record if available
+      if (!records.isEmpty()) {
+        SinkRecord firstRecord = records.iterator().next();
+        Context extractedContext =
+            KafkaConnectSingletons.propagator()
+                .extract(
+                    parentContext, firstRecord, KafkaConnectSingletons.sinkRecordHeaderGetter());
+        parentContext = extractedContext;
+      }
+
+      task = new KafkaConnectTask(records);
+      if (!KafkaConnectSingletons.instrumenter().shouldStart(parentContext, task)) {
+        return;
+      }
+
+      context = KafkaConnectSingletons.instrumenter().start(parentContext, task);
+      scope = context.makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void onExit(
+        @Advice.Thrown Throwable throwable,
+        @Advice.Local("otelTask") KafkaConnectTask task,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+      KafkaConnectSingletons.instrumenter().end(context, task, null, throwable);
+    }
+  }
+}
@@ -0,0 +1,41 @@
+plugins {
+  id("otel.java-conventions")
+  id("otel.javaagent-testing")
+}
+
+dependencies {
+  api(project(":testing-common"))
+
+  implementation("org.apache.kafka:kafka-clients:0.11.0.0")
+
+  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.postgresql:postgresql:42.7.2") // PostgreSQL JDBC driver
+  implementation("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.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("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")
+}