DataDog · ericfirth · Aug 4, 2026 · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026
@@ -62,15 +62,15 @@ jobs:
     needs:
     - build
     # If you change the following comment, update the pattern in the update_system_test_reference.sh script to match.
-    uses: DataDog/system-tests/.github/workflows/system-tests.yml@main # system tests are pinned on release branches only
+    uses: DataDog/system-tests/.github/workflows/system-tests.yml@20ec39676051d9eedf2fdcea17f716e7b89d3c10 # system tests are pinned on release branches only
     permissions:
       contents: read
       id-token: write
       packages: write
     with:
       library: java
        # If you change the following comment, update the pattern in the update_system_test_reference.sh script to match.
-      ref: "main" # system tests are pinned on release branches only
+      ref: "20ec39676051d9eedf2fdcea17f716e7b89d3c10" # system tests are pinned on release branches only
       binaries_artifact: binaries
       desired_execution_time: 900  # 15 minutes
       scenarios_groups: tracer-release

@@ -1475,7 +1475,7 @@ publishing-gate:
 
 configure_system_tests:
   variables:
-    SYSTEM_TESTS_REF: "main" # system tests are pinned on release branches only
+    SYSTEM_TESTS_REF: "20ec39676051d9eedf2fdcea17f716e7b89d3c10" # system tests are pinned on release branches only
     SYSTEM_TESTS_SCENARIOS_GROUPS: "simple_onboarding,simple_onboarding_profiling,simple_onboarding_appsec,docker-ssi,lib-injection"
 
 create_key:

@@ -1,4 +1,4 @@
 # DO NOT EDIT THIS FILE MANUALLY
 # This file is auto-generated by automation.
 include:
-  - remote: https://gitlab-templates.ddbuild.io/libdatadog/include/versions/1.1.0/one-pipeline.yml
+  - remote: https://gitlab-templates.ddbuild.io/libdatadog/include/versions/1.3.1/one-pipeline.yml
@@ -29,6 +29,7 @@ dependencies {
   compileOnly project(':products:metrics:metrics-lib')
 
   testImplementation project(':dd-java-agent:testing')
+  testImplementation project(':utils:test-junit-utils')
 }
 
 tasks.named("compileMain_java11Java", JavaCompile) {

@@ -331,6 +331,9 @@ public static Set<InstrumenterModule.TargetSystem> getEnabledSystems() {
     if (cfg.isUsmEnabled()) {
       enabledSystems.add(InstrumenterModule.TargetSystem.USM);
     }
+    if (cfg.isDataStreamsEnabled()) {
+      enabledSystems.add(InstrumenterModule.TargetSystem.DATA_STREAMS);
+    }
     if (cfg.isLlmObsEnabled()) {
       enabledSystems.add(InstrumenterModule.TargetSystem.LLMOBS);
     }

@@ -0,0 +1,75 @@
+package datadog.trace.agent.tooling;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.agent.tooling.InstrumenterModule.TargetSystem;
+import datadog.trace.test.junit.utils.config.WithConfig;
+import datadog.trace.test.junit.utils.config.WithConfigExtension;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for {@link AgentInstaller#getEnabledSystems()} to verify that it correctly includes target
+ * systems based on their corresponding configuration flags.
+ */
+@ExtendWith(WithConfigExtension.class)
+class AgentInstallerGetEnabledSystemsTest {
+
+  /**
+   * Verifies that DATA_STREAMS target system is not included when data.streams.enabled is false
+   * (default).
+   */
+  @Test
+  void dataStreamsNotIncludedWhenDisabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertFalse(
+        enabledSystems.contains(TargetSystem.DATA_STREAMS),
+        "DATA_STREAMS should not be included when disabled");
+  }
+
+  /** Verifies that DATA_STREAMS target system is included when data.streams.enabled is true. */
+  @Test
+  @WithConfig(key = "data.streams.enabled", value = "true")
+  void dataStreamsIncludedWhenEnabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertTrue(
+        enabledSystems.contains(TargetSystem.DATA_STREAMS),
+        "DATA_STREAMS should be included when enabled");
+  }
+
+  /** Verifies that USM target system is not included when usm.enabled is false (default). */
+  @Test
+  void usmNotIncludedWhenDisabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertFalse(
+        enabledSystems.contains(TargetSystem.USM), "USM should not be included when disabled");
+  }
+
+  /** Verifies that USM target system is included when usm.enabled is true. */
+  @Test
+  @WithConfig(key = "usm.enabled", value = "true")
+  void usmIncludedWhenEnabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertTrue(enabledSystems.contains(TargetSystem.USM), "USM should be included when enabled");
+  }
+
+  /** Verifies that LLMOBS target system is not included when llmobs.enabled is false (default). */
+  @Test
+  void llmobsNotIncludedWhenDisabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertFalse(
+        enabledSystems.contains(TargetSystem.LLMOBS),
+        "LLMOBS should not be included when disabled");
+  }
+
+  /** Verifies that LLMOBS target system is included when llmobs.enabled is true. */
+  @Test
+  @WithConfig(key = "llmobs.enabled", value = "true")
+  void llmobsIncludedWhenEnabled() {
+    Set<TargetSystem> enabledSystems = AgentInstaller.getEnabledSystems();
+    assertTrue(
+        enabledSystems.contains(TargetSystem.LLMOBS), "LLMOBS should be included when enabled");
+  }
+}
@@ -48,6 +48,7 @@ dependencies {
   api libs.bytebuddyagent
 
   testImplementation project(':dd-java-agent:testing')
+  testImplementation project(':utils:test-junit-utils')
   testImplementation libs.bytebuddy
   testImplementation group: 'com.google.guava', name: 'guava-testlib', version: '20.0'
 

@@ -41,6 +41,7 @@ public abstract class InstrumenterModule implements Instrumenter {
    *   <li>{@link TargetSystem#IAST iast}
    *   <li>{@link TargetSystem#CIVISIBILITY ci-visibility}
    *   <li>{@link TargetSystem#USM usm}
+   *   <li>{@link TargetSystem#DATA_STREAMS data-streams}
    *   <li>{@link TargetSystem#CONTEXT_TRACKING context-tracking}
    *   <li>{@link TargetSystem#RASP rasp}
    * </ul>
@@ -53,6 +54,7 @@ public enum TargetSystem {
     CIVISIBILITY,
     USM,
     LLMOBS,
+    DATA_STREAMS,
     CONTEXT_TRACKING,
     RASP,
   }
@@ -320,6 +322,24 @@ public final boolean isApplicable(Set<TargetSystem> enabledSystems) {
     }
   }
 
+  /** Parent class for instrumentations that support both tracing and Data Streams Monitoring */
+  public abstract static class DataStreams extends InstrumenterModule {
+    public DataStreams(String instrumentationName, String... additionalNames) {
+      super(instrumentationName, additionalNames);
+    }
+
+    @Override
+    public final boolean isApplicable(Set<TargetSystem> enabledSystems) {
+      return enabledSystems.contains(TargetSystem.TRACING)
+          || enabledSystems.contains(TargetSystem.DATA_STREAMS);
+    }
+
+    @Override
+    public boolean isEnabled() {
+      return super.isEnabled() || InstrumenterConfig.get().isDataStreamsEnabled();
+    }
+  }
+
   /** Parent class for all CI related instrumentations */
   public abstract static class CiVisibility extends InstrumenterModule {
     public CiVisibility(String instrumentationName, String... additionalNames) {

@@ -0,0 +1,102 @@
+package datadog.trace.agent.tooling;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.agent.tooling.InstrumenterModule.TargetSystem;
+import datadog.trace.test.junit.utils.config.WithConfig;
+import datadog.trace.test.junit.utils.config.WithConfigExtension;
+import java.util.HashSet;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(WithConfigExtension.class)
+class InstrumenterModuleTest {
+
+  @Test
+  void testDataStreamsIsApplicableWithTracing() {
+    Set<TargetSystem> enabledSystems = new HashSet<>();
+    enabledSystems.add(TargetSystem.TRACING);
+
+    InstrumenterModule.DataStreams module = new InstrumenterModule.DataStreams("test-module") {};
+
+    assertTrue(module.isApplicable(enabledSystems));
+  }
+
+  @Test
+  void testDataStreamsIsApplicableWithDataStreams() {
+    Set<TargetSystem> enabledSystems = new HashSet<>();
+    enabledSystems.add(TargetSystem.DATA_STREAMS);
+
+    InstrumenterModule.DataStreams module = new InstrumenterModule.DataStreams("test-module") {};
+
+    assertTrue(module.isApplicable(enabledSystems));
+  }
+
+  @Test
+  void testDataStreamsIsApplicableWithBoth() {
+    Set<TargetSystem> enabledSystems = new HashSet<>();
+    enabledSystems.add(TargetSystem.TRACING);
+    enabledSystems.add(TargetSystem.DATA_STREAMS);
+
+    InstrumenterModule.DataStreams module = new InstrumenterModule.DataStreams("test-module") {};
+
+    assertTrue(module.isApplicable(enabledSystems));
+  }
+
+  @Test
+  void testDataStreamsIsApplicableWithNeither() {
+    Set<TargetSystem> enabledSystems = new HashSet<>();
+    enabledSystems.add(TargetSystem.APPSEC);
+
+    InstrumenterModule.DataStreams module = new InstrumenterModule.DataStreams("test-module") {};
+
+    assertFalse(module.isApplicable(enabledSystems));
+  }
+
+  @Test
+  @WithConfig(key = "trace.test-kafka-module.enabled", value = "false")
+  @WithConfig(key = "data.streams.enabled", value = "true")
+  void testDataStreamsIsEnabledWhenDataStreamsEnabledOverridesFalse() {
+    // When tracing for this integration is disabled but DSM is explicitly enabled,
+    // isEnabled() should still return true.
+    InstrumenterModule.DataStreams module =
+        new InstrumenterModule.DataStreams("test-kafka-module") {};
+
+    assertTrue(module.isEnabled());
+  }
+
+  @Test
+  @WithConfig(key = "trace.test-kafka-module.enabled", value = "true")
+  @WithConfig(key = "data.streams.enabled", value = "false")
+  void testDataStreamsIsEnabledWhenSuperEnabledIsTrue() {
+    // When super.isEnabled() is true, isEnabled() should return true regardless of DSM state.
+    InstrumenterModule.DataStreams module =
+        new InstrumenterModule.DataStreams("test-kafka-module") {};
+
+    assertTrue(module.isEnabled());
+  }
+
+  @Test
+  @WithConfig(key = "trace.test-kafka-module.enabled", value = "true")
+  @WithConfig(key = "data.streams.enabled", value = "true")
+  void testDataStreamsIsEnabledWhenBothEnabled() {
+    // When both super.isEnabled() and DSM are enabled, isEnabled() should return true.
+    InstrumenterModule.DataStreams module =
+        new InstrumenterModule.DataStreams("test-kafka-module") {};
+
+    assertTrue(module.isEnabled());
+  }
+
+  @Test
+  @WithConfig(key = "trace.test-kafka-module.enabled", value = "false")
+  @WithConfig(key = "data.streams.enabled", value = "false")
+  void testDataStreamsIsEnabledWhenBothDisabled() {
+    // When both super.isEnabled() and DSM are disabled, isEnabled() should return false.
+    InstrumenterModule.DataStreams module =
+        new InstrumenterModule.DataStreams("test-kafka-module") {};
+
+    assertFalse(module.isEnabled());
+  }
+}
@@ -5,7 +5,7 @@
 ch.qos.logback:logback-classic:1.2.13=testCompileClasspath,testRuntimeClasspath
 ch.qos.logback:logback-core:1.2.13=testCompileClasspath,testRuntimeClasspath
 com.datadoghq:dd-javac-plugin-client:0.2.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
-com.datadoghq:ddprof:1.48.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+com.datadoghq:ddprof:1.49.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 com.github.javaparser:javaparser-core:3.25.6=codenarc
 com.github.spotbugs:spotbugs-annotations:4.9.8=compileClasspath,spotbugs
 com.github.spotbugs:spotbugs:4.9.8=spotbugs

@@ -23,11 +23,11 @@
 import org.apache.kafka.common.TopicPartition;
 
 @AutoService(InstrumenterModule.class)
-public final class ConsumerCoordinatorInstrumentation extends InstrumenterModule.Tracing
+public final class ConsumerCoordinatorInstrumentation extends InstrumenterModule.DataStreams
     implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
 
   public ConsumerCoordinatorInstrumentation() {
-    super("kafka", "kafka-0.11");
+    super(KafkaDecorator.INTEGRATION_NAME, KafkaDecorator.LEGACY_INTEGRATION_NAME);
   }
 
   @Override

@@ -44,11 +44,11 @@
  * and cluster ID, in the context store for later use.
  */
 @AutoService(InstrumenterModule.class)
-public final class KafkaConsumerInfoInstrumentation extends InstrumenterModule.Tracing
+public final class KafkaConsumerInfoInstrumentation extends InstrumenterModule.DataStreams
     implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
 
   public KafkaConsumerInfoInstrumentation() {
-    super("kafka", "kafka-0.11");
+    super(KafkaDecorator.INTEGRATION_NAME, KafkaDecorator.LEGACY_INTEGRATION_NAME);
   }
 
   @Override
@@ -252,7 +252,10 @@ public static AgentScope onEnter(@Advice.This KafkaConsumer consumer) {
         }
       }
 
-      if (traceConfig().isDataStreamsEnabled()) {
+      if (traceConfig().isDataStreamsEnabled() && KafkaDecorator.TRACING_ENABLED) {
+        // DSM-only mode (tracing disabled) never creates a real poll span: TracingIterator
+        // carries its pathway context on a lightweight, never-collected span shim instead, so
+        // there's nothing here that needs wrapping/protecting from being force-dropped.
         final AgentSpan span = startSpan(JAVA_KAFKA.toString(), KAFKA_POLL);
         return activateSpan(span);
       }

@@ -27,11 +27,11 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 @AutoService(InstrumenterModule.class)
-public final class KafkaConsumerInstrumentation extends InstrumenterModule.Tracing
+public final class KafkaConsumerInstrumentation extends InstrumenterModule.DataStreams
     implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
 
   public KafkaConsumerInstrumentation() {
-    super("kafka", "kafka-0.11");
+    super(KafkaDecorator.INTEGRATION_NAME, KafkaDecorator.LEGACY_INTEGRATION_NAME);
   }
 
   @Override

@@ -12,6 +12,7 @@
 
 import datadog.trace.api.Config;
 import datadog.trace.api.Functions;
+import datadog.trace.api.InstrumenterConfig;
 import datadog.trace.api.cache.DDCache;
 import datadog.trace.api.cache.DDCaches;
 import datadog.trace.api.naming.SpanNaming;
@@ -20,6 +21,7 @@
 import datadog.trace.bootstrap.instrumentation.api.Tags;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.bootstrap.instrumentation.decorator.MessagingClientDecorator;
+import java.util.Arrays;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -29,6 +31,10 @@
 
 public class KafkaDecorator extends MessagingClientDecorator {
   private static final String KAFKA = "kafka";
+  // Kept in sync with the names each kafka-clients-0.11 instrumentation module passes to its own
+  // super(...) constructor call, so TRACING_ENABLED can't drift from what is actually registered.
+  public static final String INTEGRATION_NAME = KAFKA;
+  public static final String LEGACY_INTEGRATION_NAME = "kafka-0.11";
   public static final CharSequence JAVA_KAFKA = UTF8BytesString.create("java-kafka");
   public static final CharSequence KAFKA_CONSUME =
       UTF8BytesString.create(
@@ -42,6 +48,11 @@ public class KafkaDecorator extends MessagingClientDecorator {
   public static final boolean KAFKA_LEGACY_TRACING = Config.get().isKafkaLegacyTracingEnabled();
   public static final boolean TIME_IN_QUEUE_ENABLED =
       Config.get().isTimeInQueueEnabled(!KAFKA_LEGACY_TRACING, KAFKA);
+  public static final boolean TRACING_ENABLED =
+      InstrumenterConfig.get()
+          .isIntegrationEnabled(
+              Arrays.asList(INTEGRATION_NAME, LEGACY_INTEGRATION_NAME),
+              InstrumenterConfig.get().isIntegrationsEnabled());
   public static final String KAFKA_PRODUCED_KEY = "x_datadog_kafka_produced";
   private final String spanKind;
   private final CharSequence spanType;