wip: instrumenting kafka connect.

Author: vasantteja

instrumentation/kafka/kafka-connect-2.6/javaagent/README.md

@@ -0,0 +1,101 @@
+# Kafka Connect SinkTask Integration Test
+
+This module contains integration tests for Kafka Connect SinkTask instrumentation using Testcontainers.
+
+## Test Overview
+
+The `KafkaConnectSinkTaskTest` class demonstrates how to test Kafka Connect SinkTask functionality with:
+
+- **Kafka Container**: Apache Kafka using Confluent Platform
+- **PostgreSQL Container**: PostgreSQL database as the sink destination
+- **Kafka Connect Container**: Confluent Kafka Connect with JDBC connector
+
+## Test Structure
+
+### Test Setup
+1. **Kafka Container**: Starts a Kafka broker for message streaming
+2. **PostgreSQL Container**: Starts a PostgreSQL database for data storage
+3. **Kafka Connect Container**: Starts Kafka Connect with the Confluent JDBC connector pre-installed
+4. **Network Configuration**: All containers are connected via a Docker network for inter-container communication
+
+### Test Scenarios
+
+#### 1. Basic SinkTask Test (`testKafkaConnectSinkTask`)
+- Sends JSON messages to a Kafka topic
+- Verifies that Kafka Connect processes the messages and inserts them into PostgreSQL
+- Validates both the count and content of inserted records
+
+#### 2. Batch Processing Test (`testKafkaConnectSinkTaskWithMultipleBatches`)
+- Sends multiple batches of messages to test bulk processing
+- Verifies that all messages are correctly processed and stored
+
+### Data Flow
+```
+JSON Messages → Kafka Topic → Kafka Connect JDBC Sink → PostgreSQL Table
+```
+
+## Running the Tests
+
+### Prerequisites
+- Docker installed and running
+- Java 11 or higher
+- Gradle
+
+### Execute Tests
+```bash
+# Run all tests in the module
+./gradlew :instrumentation:kafka:kafka-connect-2.6:javaagent:test
+
+# Run only the SinkTask test
+./gradlew :instrumentation:kafka:kafka-connect-2.6:javaagent:test --tests "KafkaConnectSinkTaskTest"
+```
+
+### Test Configuration
+
+The test uses the following configuration:
+- **Kafka Topic**: `test-topic`
+- **PostgreSQL Database**: `testdb`
+- **Table Name**: `test_table`
+- **Connector**: Confluent JDBC Sink Connector (version 10.7.4)
+
+### Test Data Format
+
+Messages are sent as JSON strings with the following structure:
+```json
+{
+  "name": "John Doe",
+  "value": 100
+}
+```
+
+These are automatically mapped to PostgreSQL table columns:
+- `id` (SERIAL PRIMARY KEY)
+- `name` (VARCHAR(255))
+- `value` (INTEGER)
+- `created_at` (TIMESTAMP)
+
+## Key Features
+
+1. **Full Integration**: Tests the complete data pipeline from Kafka to PostgreSQL
+2. **Container Orchestration**: Uses Docker networks for seamless container communication
+3. **Realistic Environment**: Uses actual Confluent components, not mocks
+4. **Data Validation**: Verifies both data count and content accuracy
+5. **Error Handling**: Includes proper cleanup and error handling
+6. **Batch Processing**: Tests both single messages and batch processing scenarios
+
+## Dependencies
+
+The test relies on the following key dependencies:
+- `org.testcontainers:kafka` - Kafka container support
+- `org.testcontainers:postgresql` - PostgreSQL container support
+- `org.postgresql:postgresql` - PostgreSQL JDBC driver
+- `com.fasterxml.jackson.core:jackson-databind` - JSON processing
+- `org.awaitility:awaitility` - Asynchronous testing utilities
+
+## Notes
+
+- The test automatically installs the Confluent JDBC connector during container startup
+- All containers are started with appropriate logging for debugging
+- The test includes proper timeouts and retry logic for container readiness
+- Network isolation ensures tests don't interfere with each other
+- The test focuses on data flow verification rather than OpenTelemetry spans (as requested)

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

@@ -0,0 +1,41 @@
+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:3.6.1")
+
+  testImplementation("org.apache.kafka:connect-runtime:3.6.1")
+
+}
+
+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")
+  }
+}

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

@@ -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);
+    }
+  }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+import java.util.logging.Logger;
+
+@AutoService(InstrumentationModule.class)
+public class KafkaConnectInstrumentationModule extends InstrumentationModule {
+
+  private static final Logger logger = Logger.getLogger(KafkaConnectInstrumentationModule.class.getName());
+
+  public KafkaConnectInstrumentationModule() {
+    super("kafka-connect", "kafka-connect-2.6");
+    logger.info("KafkaConnect: InstrumentationModule constructor called");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    logger.info("KafkaConnect: typeInstrumentations() called");
+    return singletonList(new SinkTaskInstrumentation());
+  }
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.TextMapPropagator;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+
+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 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;
+  }
+
+  private KafkaConnectSingletons() {}
+}

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

@@ -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";
+  }
+}

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

@@ -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();
+  }
+}