wip: removing the parent child relationship between kafka producer and kafka connect spans.

Author: vasantteja

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

@@ -46,18 +46,11 @@ public static void onEnter(
         @Advice.Local("otelContext") Context context,
         @Advice.Local("otelScope") Scope scope) {
 
+      // For batch processing from multiple traces, use root context instead of extracting
+      // from first record to avoid creating incorrect parent-child relationships.
+      // Relationships are maintained through span links (see KafkaConnectBatchProcessSpanLinksExtractor)
       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 (!instrumenter().shouldStart(parentContext, task)) {
         return;

instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/MongoKafkaConnectSinkTaskTest.java

@@ -360,6 +360,18 @@ public void testKafkaConnectMongoSinkTaskInstrumentation()
     String tracesJson =
         given().when().get(backendUrl + "/get-traces").then().statusCode(200).extract().asString();
 
+    // Write resourceSpans to desktop file for inspection
+    try {
+      java.nio.file.Path desktopPath =
+          java.nio.file.Paths.get(
+              System.getProperty("user.home"), "Desktop", "kafka-connect-single-topic-spans.json");
+      java.nio.file.Files.write(
+          desktopPath, tracesJson.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+      System.out.println("✅ Wrote resourceSpans to: " + desktopPath.toString());
+    } catch (Exception e) {
+      System.err.println("❌ Failed to write spans to desktop: " + e.getMessage());
+    }
+
     // Extract spans and links using separate deserialization method
     TracingData tracingData;
     try {
@@ -405,6 +417,11 @@ public void testKafkaConnectMongoSinkTaskInstrumentation()
         .isNotNull()
         .as("Database span parent should be Kafka Connect Consumer span")
         .isEqualTo(tracingData.kafkaConnectConsumerSpan.spanId);
+
+    // Assertion 5: Verify Kafka Connect Consumer span has no parent (batch processing pattern)
+    assertThat(tracingData.kafkaConnectConsumerSpan.parentSpanId)
+        .as("Kafka Connect Consumer span should have no parent for batch processing from multiple traces")
+        .isNull();
   }
 
   @Test
@@ -534,6 +551,11 @@ public void testKafkaConnectMongoSinkTaskMultiTopicInstrumentation()
         .isNotNull()
         .as("Database span parent should be Kafka Connect Consumer span")
         .isEqualTo(tracingData.kafkaConnectConsumerSpan.spanId);
+
+    // Assertion 7: Verify Kafka Connect Consumer span has no parent (batch processing pattern)
+    assertThat(tracingData.kafkaConnectConsumerSpan.parentSpanId)
+        .as("Kafka Connect Consumer span should have no parent for batch processing from multiple traces")
+        .isNull();
   }
 
   // Private methods

instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/PostgresKafkaConnectSinkTaskTest.java

@@ -386,6 +386,18 @@ public void testKafkaConnectPostgresSinkTaskInstrumentation()
     String tracesJson =
         given().when().get(backendUrl + "/get-traces").then().statusCode(200).extract().asString();
 
+    // Write resourceSpans to desktop file for inspection
+    try {
+      java.nio.file.Path desktopPath =
+          java.nio.file.Paths.get(
+              System.getProperty("user.home"), "Desktop", "kafka-connect-postgres-spans.json");
+      java.nio.file.Files.write(
+          desktopPath, tracesJson.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+      System.out.println("✅ Wrote resourceSpans to: " + desktopPath.toString());
+    } catch (Exception e) {
+      System.err.println("❌ Failed to write spans to desktop: " + e.getMessage());
+    }
+
     // Extract spans and links using separate deserialization method
     TracingData tracingData;
     try {
@@ -431,6 +443,11 @@ public void testKafkaConnectPostgresSinkTaskInstrumentation()
         .isNotNull()
         .as("Database span parent should be Kafka Connect Consumer span")
         .isEqualTo(tracingData.kafkaConnectConsumerSpan.spanId);
+
+    // Assertion 5: Verify Kafka Connect Consumer span has no parent (batch processing pattern)
+    assertThat(tracingData.kafkaConnectConsumerSpan.parentSpanId)
+        .as("Kafka Connect Consumer span should have no parent for batch processing from multiple traces")
+        .isNull();
   }
 
   @AfterAll
@@ -661,7 +678,7 @@ private static TracingData deserializeAndExtractSpans(String tracesJson, String
                 && spanName.contains(expectedTopicName)
                 && spanKind.equals("SPAN_KIND_CONSUMER")) {
               kafkaConnectConsumerSpan =
-                  new SpanInfo(spanName, traceId, spanId, parentSpanId, spanKind);
+                  new SpanInfo(traceId, spanId, parentSpanId, spanKind);
 
               // Extract span link information for verification
               JsonNode linksArray = spanNode.get("links");
@@ -681,7 +698,7 @@ private static TracingData deserializeAndExtractSpans(String tracesJson, String
                     || spanName.contains("SELECT")
                     || spanName.contains(DB_TABLE_PERSON))) {
               databaseSpan =
-                  new SpanInfo(spanName, traceId, spanId, parentSpanId, spanKind);
+                  new SpanInfo(traceId, spanId, parentSpanId, spanKind);
             }
           }
         }
@@ -716,7 +733,6 @@ private static class SpanInfo {
     final String kind;
 
     SpanInfo(
-        String name,
         String traceId,
         String spanId,
         String parentSpanId,