Add a new configuration kafka.client.propagation.extraction.enabled to make this behavior optional

Author: PlugaruT

dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerInstrumentation.java

@@ -127,9 +127,11 @@ public static AgentScope onEnter(
         ClusterIdHolder.set(clusterId);
       }
 
-      // Try to extract existing trace context from record headers
+      // Try to extract existing trace context from record headers if enabled
       final AgentSpanContext extractedContext =
-          extractContextAndGetSpanContext(record.headers(), TextMapExtractAdapter.GETTER);
+          Config.get().isKafkaClientPropagationExtractionEnabled()
+              ? extractContextAndGetSpanContext(record.headers(), TextMapExtractAdapter.GETTER)
+              : null;
 
       final AgentSpan localActiveSpan = activeSpan();
 

dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy

@@ -1024,6 +1024,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
 
   def "test producer extracts and uses existing trace context from record headers"() {
     setup:
+    injectSysConfig("kafka.client.propagation.extraction.enabled", "true")
     def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString())
     def producer = new KafkaProducer<>(senderProps)
 
@@ -1052,6 +1053,35 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     producer?.close()
   }
 
+  def "test producer ignores existing trace context when extraction is disabled"() {
+    setup:
+    injectSysConfig("kafka.client.propagation.extraction.enabled", "false")
+    def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString())
+    def producer = new KafkaProducer<>(senderProps)
+
+    def existingTraceId = 1234567890123456L
+    def existingSpanId = 9876543210987654L
+    def headers = new RecordHeaders()
+    headers.add(new RecordHeader("x-datadog-trace-id",
+      String.valueOf(existingTraceId).getBytes(StandardCharsets.UTF_8)))
+    headers.add(new RecordHeader("x-datadog-parent-id",
+      String.valueOf(existingSpanId).getBytes(StandardCharsets.UTF_8)))
+
+    when:
+    def record = new ProducerRecord(SHARED_TOPIC, 0, null, "test-context-extraction-disabled", headers)
+    producer.send(record).get()
+
+    then:
+    TEST_WRITER.waitForTraces(1)
+    def producedSpan = TEST_WRITER[0][0]
+    // Verify the span did NOT use the extracted context as parent
+    producedSpan.traceId.toLong() != existingTraceId
+    producedSpan.parentId != existingSpanId
+
+    cleanup:
+    producer?.close()
+  }
+
   def containerProperties() {
     try {
       // Different class names for test and latestDepTest.

dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ProducerAdvice.java

@@ -48,9 +48,11 @@ public static AgentScope onEnter(
       ClusterIdHolder.set(clusterId);
     }
 
-    // Try to extract existing trace context from record headers
+    // Try to extract existing trace context from record headers if enabled
     final AgentSpanContext extractedContext =
-        extractContextAndGetSpanContext(record.headers(), TextMapExtractAdapter.GETTER);
+        Config.get().isKafkaClientPropagationExtractionEnabled()
+            ? extractContextAndGetSpanContext(record.headers(), TextMapExtractAdapter.GETTER)
+            : null;
 
     final AgentSpan localActiveSpan = activeSpan();
 

dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy

@@ -859,6 +859,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
 
   def "test producer extracts and uses existing trace context from record headers"() {
     setup:
+    injectSysConfig("kafka.client.propagation.extraction.enabled", "true")
     def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString())
     def producer = new KafkaProducer<>(senderProps)
 
@@ -887,6 +888,35 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     producer?.close()
   }
 
+  def "test producer ignores existing trace context when extraction is disabled"() {
+    setup:
+    injectSysConfig("kafka.client.propagation.extraction.enabled", "false")
+    def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString())
+    def producer = new KafkaProducer<>(senderProps)
+
+    def existingTraceId = 1234567890123456L
+    def existingSpanId = 9876543210987654L
+    def headers = new RecordHeaders()
+    headers.add(new RecordHeader("x-datadog-trace-id",
+      String.valueOf(existingTraceId).getBytes(StandardCharsets.UTF_8)))
+    headers.add(new RecordHeader("x-datadog-parent-id",
+      String.valueOf(existingSpanId).getBytes(StandardCharsets.UTF_8)))
+
+    when:
+    def record = new ProducerRecord(SHARED_TOPIC, 0, null, "test-context-extraction-disabled", headers)
+    producer.send(record).get()
+
+    then:
+    TEST_WRITER.waitForTraces(1)
+    def producedSpan = TEST_WRITER[0][0]
+    // Verify the span did NOT use the extracted context as parent
+    producedSpan.traceId.toLong() != existingTraceId
+    producedSpan.parentId != existingSpanId
+
+    cleanup:
+    producer?.close()
+  }
+
   def producerSpan(
     TraceAssert trace,
     Map<String, ?> config,

dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java

@@ -104,6 +104,8 @@ public final class TraceInstrumentationConfig {
       "kafka.client.propagation.disabled.topics";
   public static final String KAFKA_CLIENT_BASE64_DECODING_ENABLED =
       "kafka.client.base64.decoding.enabled";
+  public static final String KAFKA_CLIENT_PROPAGATION_EXTRACTION_ENABLED =
+      "kafka.client.propagation.extraction.enabled";
 
   public static final String JMS_PROPAGATION_DISABLED_TOPICS = "jms.propagation.disabled.topics";
   public static final String JMS_PROPAGATION_DISABLED_QUEUES = "jms.propagation.disabled.queues";

internal-api/src/main/java/datadog/trace/api/Config.java

@@ -562,6 +562,7 @@
 import static datadog.trace.api.config.TraceInstrumentationConfig.JMS_UNACKNOWLEDGED_MAX_AGE;
 import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_BASE64_DECODING_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS;
+import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_PROPAGATION_EXTRACTION_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION;
 import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.MESSAGE_BROKER_SPLIT_BY_DESTINATION;
@@ -1158,6 +1159,7 @@ public static String getHostName() {
   private final boolean kafkaClientPropagationEnabled;
   private final Set<String> kafkaClientPropagationDisabledTopics;
   private final boolean kafkaClientBase64DecodingEnabled;
+  private final boolean kafkaClientPropagationExtractionEnabled;
 
   private final boolean jmsPropagationEnabled;
   private final Set<String> jmsPropagationDisabledTopics;
@@ -2629,6 +2631,8 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment())
         tryMakeImmutableSet(configProvider.getList(KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS));
     kafkaClientBase64DecodingEnabled =
         configProvider.getBoolean(KAFKA_CLIENT_BASE64_DECODING_ENABLED, false);
+    kafkaClientPropagationExtractionEnabled =
+        configProvider.getBoolean(KAFKA_CLIENT_PROPAGATION_EXTRACTION_ENABLED, false);
     jmsPropagationEnabled = isPropagationEnabled(true, "jms");
     jmsPropagationDisabledTopics =
         tryMakeImmutableSet(configProvider.getList(JMS_PROPAGATION_DISABLED_TOPICS));
@@ -4451,6 +4455,10 @@ public boolean isKafkaClientBase64DecodingEnabled() {
     return kafkaClientBase64DecodingEnabled;
   }
 
+  public boolean isKafkaClientPropagationExtractionEnabled() {
+    return kafkaClientPropagationExtractionEnabled;
+  }
+
   public boolean isRabbitPropagationEnabled() {
     return rabbitPropagationEnabled;
   }
@@ -5952,6 +5960,8 @@ public String toString() {
         + kafkaClientPropagationDisabledTopics
         + ", kafkaClientBase64DecodingEnabled="
         + kafkaClientBase64DecodingEnabled
+        + ", kafkaClientPropagationExtractionEnabled="
+        + kafkaClientPropagationExtractionEnabled
         + ", jmsPropagationEnabled="
         + jmsPropagationEnabled
         + ", jmsPropagationDisabledTopics="