Merge branch 'master' into mhlidd/add_logback_config

Author: mhlidd

.github/dependabot.yml

@@ -1,36 +1,38 @@
+# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
 version: 2
 updates:
-- package-ecosystem: github-actions
-  directory: /
-  schedule:
-    interval: weekly
-  labels:
-  - 'comp: tooling'
-  - 'tag: dependencies'
-  - 'tag: no release notes'
-  commit-message:
-    prefix: 'chore(ci): '
-  groups:
-    gh-actions-packages:
-      patterns:
-      - '*'
-  cooldown:
-    default-days: 14
-- package-ecosystem: gradle
-  directory: /
-  schedule:
-    interval: weekly
-  allow:
-  - dependency-name: gradle
-  ignore:
-  - dependency-name: gradle
-    update-types:
-    - version-update:semver-major
-  labels:
-  - 'comp: tooling'
-  - 'tag: dependencies'
-  - 'tag: no release notes'
-  commit-message:
-    prefix: 'chore(build): '
-  cooldown:
-    default-days: 14
+  - package-ecosystem: github-actions
+    directory: /
+    schedule:
+      interval: weekly
+    labels:
+      - 'comp: tooling'
+      - 'tag: dependencies'
+      - 'tag: no release notes'
+    commit-message:
+      prefix: 'chore(ci): '
+    groups:
+      gh-actions-packages:
+        patterns:
+          - '*'
+    cooldown:
+      default-days: 2
+
+  - package-ecosystem: gradle
+    directory: /
+    schedule:
+      interval: weekly
+    allow:
+      - dependency-name: gradle
+    ignore:
+      - dependency-name: gradle
+        update-types:
+          - version-update:semver-major
+    labels:
+      - 'comp: tooling'
+      - 'tag: dependencies'
+      - 'tag: no release notes'
+    commit-message:
+      prefix: 'chore(build): '
+    cooldown:
+      default-days: 2

.github/workflows/analyze-changes.yaml

@@ -20,7 +20,7 @@ jobs:
         with:
           submodules: 'recursive'
       - name: Cache Gradle dependencies
-        uses: actions/cache@cdf6c1fa76f9f475f3d7449005a359c84ca0f306 # v5.0.3
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
         with:
           path: |
             ~/.gradle/caches
@@ -30,7 +30,7 @@ jobs:
             ${{ runner.os }}-gradle-
         
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@b1bff81932f5cdfc8695c7752dcee935dcd061c8 # v4.33.0
+        uses: github/codeql-action/init@c10b8064de6f491fea524254123dbe5e09572f13 # v4.35.1
         with:
           languages: 'java'
           build-mode: 'manual'
@@ -43,7 +43,7 @@ jobs:
           ./gradlew clean :dd-java-agent:shadowJar --build-cache --parallel --stacktrace --no-daemon --max-workers=4
 
       - name: Perform CodeQL Analysis and upload results to GitHub Security tab
-        uses: github/codeql-action/analyze@b1bff81932f5cdfc8695c7752dcee935dcd061c8 # v4.33.0
+        uses: github/codeql-action/analyze@c10b8064de6f491fea524254123dbe5e09572f13 # v4.35.1
 
   trivy:
     name: Analyze changes with Trivy
@@ -60,7 +60,7 @@ jobs:
           submodules: 'recursive'
 
       - name: Cache Gradle dependencies
-        uses: actions/cache@cdf6c1fa76f9f475f3d7449005a359c84ca0f306 # v5.0.3
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
         with:
           path: |
             ~/.gradle/caches
@@ -102,7 +102,7 @@ jobs:
           TRIVY_JAVA_DB_REPOSITORY: ghcr.io/aquasecurity/trivy-java-db,public.ecr.aws/aquasecurity/trivy-java-db
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@b1bff81932f5cdfc8695c7752dcee935dcd061c8 # v4.33.0
+        uses: github/codeql-action/upload-sarif@c10b8064de6f491fea524254123dbe5e09572f13 # v4.35.1
         if: always()
         with:
           sarif_file: 'trivy-results.sarif'

.github/workflows/run-system-tests.yaml

@@ -30,7 +30,7 @@ jobs:
           fetch-depth: 0
 
       - name: Cache Gradle dependencies
-        uses: actions/cache@cdf6c1fa76f9f475f3d7449005a359c84ca0f306 # v5.0.3
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
         with:
           path: |
             ~/.gradle/caches

dd-java-agent/build.gradle

@@ -19,8 +19,16 @@ configurations {
 def includedAgentDir = project.layout.buildDirectory.dir("generated/included")
 def includedJarFileTree = fileTree(includedAgentDir)
 
+def pomPropertiesDir = project.layout.buildDirectory.dir("generated/maven-metadata")
+def pomPropertiesFileTree = fileTree(pomPropertiesDir)
+
 tasks.named("processResources") {
   dependsOn(includedJarFileTree)
+  dependsOn(pomPropertiesFileTree)
+}
+
+tasks.named("sourcesJar") {
+  dependsOn(pomPropertiesFileTree)
 }
 
 sourceSets {
@@ -33,7 +41,7 @@ sourceSets {
   "main_java11" {
     java.srcDirs "${project.projectDir}/src/main/java11"
   }
-  main.resources.srcDir(includedAgentDir)
+  main.resources.srcDirs(includedAgentDir, pomPropertiesDir)
 }
 
 def java6CompileTask = tasks.named("compileMain_java6Java") {
@@ -324,6 +332,14 @@ def generateAgentJarIndex = tasks.register('generateAgentJarIndex', JavaExec) {
 }
 sourceSets.main.resources.srcDir(generateAgentJarIndex)
 
+def generatePomProperties = tasks.register('generatePomProperties', WriteProperties) {
+  destinationFile = pomPropertiesDir.map { it.file("META-INF/maven/com.datadoghq/dd-java-agent/pom.properties") }
+  property("groupId", "com.datadoghq")
+  property("artifactId", "dd-java-agent")
+  property("version", project.providers.provider { project.version.toString() })
+}
+pomPropertiesFileTree.builtBy(generatePomProperties)
+
 subprojects { Project subProj ->
   // Don't need javadoc task run for internal projects.
   subProj.tasks.withType(Javadoc).configureEach { enabled = false }

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

@@ -10,6 +10,7 @@
 import datadog.trace.api.datastreams.DataStreamsTags;
 import datadog.trace.bootstrap.InstrumentationContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import datadog.trace.instrumentation.kafka_common.MetadataState;
 import java.util.HashMap;
 import java.util.Map;
 import net.bytebuddy.asm.Advice;
@@ -42,7 +43,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
   @Override
   public Map<String, String> contextStore() {
     Map<String, String> contextStores = new HashMap<>();
-    contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
+    contextStores.put(
+        "org.apache.kafka.clients.Metadata",
+        "datadog.trace.instrumentation.kafka_common.MetadataState");
     contextStores.put(
         "org.apache.kafka.clients.consumer.internals.ConsumerCoordinator",
         KafkaConsumerInfo.class.getName());
@@ -56,7 +59,11 @@ public String instrumentedType() {
 
   @Override
   public String[] helperClassNames() {
-    return new String[] {packageName + ".KafkaConsumerInfo"};
+    return new String[] {
+      packageName + ".KafkaConsumerInfo",
+      "datadog.trace.instrumentation.kafka_common.PendingConfig",
+      "datadog.trace.instrumentation.kafka_common.MetadataState",
+    };
   }
 
   @Override
@@ -90,7 +97,9 @@ public static void trackCommitOffset(
       Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
       String clusterId = null;
       if (consumerMetadata != null) {
-        clusterId = InstrumentationContext.get(Metadata.class, String.class).get(consumerMetadata);
+        MetadataState metadataState =
+            InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
+        clusterId = metadataState != null ? metadataState.clusterId : null;
       }
 
       for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {

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

@@ -23,6 +23,8 @@
 import datadog.trace.bootstrap.instrumentation.api.AgentScope;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.instrumentation.kafka_common.ClusterIdHolder;
+import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper;
+import datadog.trace.instrumentation.kafka_common.MetadataState;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -60,7 +62,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
   @Override
   public Map<String, String> contextStore() {
     Map<String, String> contextStores = new HashMap<>();
-    contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
+    contextStores.put(
+        "org.apache.kafka.clients.Metadata",
+        "datadog.trace.instrumentation.kafka_common.MetadataState");
     contextStores.put(
         "org.apache.kafka.clients.consumer.ConsumerRecords", KafkaConsumerInfo.class.getName());
     contextStores.put(
@@ -82,6 +86,9 @@ public String[] helperClassNames() {
       packageName + ".KafkaDecorator",
       packageName + ".KafkaConsumerInfo",
       "datadog.trace.instrumentation.kafka_common.ClusterIdHolder",
+      "datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
+      "datadog.trace.instrumentation.kafka_common.PendingConfig",
+      "datadog.trace.instrumentation.kafka_common.MetadataState",
     };
   }
 
@@ -148,6 +155,16 @@ public static void captureGroup(
               .put(coordinator, kafkaConsumerInfo);
         }
       }
+
+      if (Config.get().isDataStreamsEnabled()) {
+        MetadataState state =
+            InstrumentationContext.get(Metadata.class, MetadataState.class)
+                .putIfAbsent(metadata, MetadataState::new);
+        KafkaConfigHelper.storePendingConsumerConfig(
+            state,
+            normalizedConsumerGroup,
+            KafkaConfigHelper.extractConsumerConfig(consumerConfig));
+      }
     }
 
     public static void muzzleCheck(ConsumerRecord record) {
@@ -191,6 +208,16 @@ public static void captureGroup(
               .put(coordinator, kafkaConsumerInfo);
         }
       }
+
+      if (Config.get().isDataStreamsEnabled()) {
+        MetadataState state =
+            InstrumentationContext.get(Metadata.class, MetadataState.class)
+                .putIfAbsent(metadata, MetadataState::new);
+        KafkaConfigHelper.storePendingConsumerConfig(
+            state,
+            normalizedConsumerGroup,
+            KafkaConfigHelper.extractConsumerConfigFromMap(consumerConfig));
+      }
     }
 
     public static void muzzleCheck(ConsumerRecord record) {
@@ -214,8 +241,9 @@ public static AgentScope onEnter(@Advice.This KafkaConsumer consumer) {
       if (kafkaConsumerInfo != null && Config.get().isDataStreamsEnabled()) {
         Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
         if (consumerMetadata != null) {
-          String clusterId =
-              InstrumentationContext.get(Metadata.class, String.class).get(consumerMetadata);
+          MetadataState metadataState =
+              InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
+          String clusterId = metadataState != null ? metadataState.clusterId : null;
           if (clusterId != null) {
             ClusterIdHolder.set(clusterId);
           }

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

@@ -15,6 +15,7 @@
 import datadog.trace.agent.tooling.Instrumenter;
 import datadog.trace.agent.tooling.InstrumenterModule;
 import datadog.trace.bootstrap.InstrumentationContext;
+import datadog.trace.instrumentation.kafka_common.MetadataState;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -46,7 +47,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
   @Override
   public Map<String, String> contextStore() {
     Map<String, String> contextStores = new HashMap<>(2);
-    contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
+    contextStores.put(
+        "org.apache.kafka.clients.Metadata",
+        "datadog.trace.instrumentation.kafka_common.MetadataState");
     contextStores.put(
         "org.apache.kafka.clients.consumer.ConsumerRecords", KafkaConsumerInfo.class.getName());
     return contextStores;
@@ -73,6 +76,8 @@ public String[] helperClassNames() {
       packageName + ".TextMapInjectAdapter",
       "datadog.trace.instrumentation.kafka_common.Utils",
       "datadog.trace.instrumentation.kafka_common.StreamingContext",
+      "datadog.trace.instrumentation.kafka_common.PendingConfig",
+      "datadog.trace.instrumentation.kafka_common.MetadataState",
     };
   }
 
@@ -113,7 +118,7 @@ public static void wrap(
         String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
         String clusterId =
             KafkaConsumerInstrumentationHelper.extractClusterId(
-                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
+                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
         String bootstrapServers =
             KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
         iterable =
@@ -135,7 +140,7 @@ public static void wrap(
         String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
         String clusterId =
             KafkaConsumerInstrumentationHelper.extractClusterId(
-                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
+                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
         String bootstrapServers =
             KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
         iterable =
@@ -157,7 +162,7 @@ public static void wrap(
         String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
         String clusterId =
             KafkaConsumerInstrumentationHelper.extractClusterId(
-                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
+                kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
         String bootstrapServers =
             KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
         iterator =

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

@@ -2,6 +2,7 @@
 
 import datadog.trace.api.Config;
 import datadog.trace.bootstrap.ContextStore;
+import datadog.trace.instrumentation.kafka_common.MetadataState;
 import org.apache.kafka.clients.Metadata;
 
 public class KafkaConsumerInstrumentationHelper {
@@ -13,11 +14,13 @@ public static String extractGroup(KafkaConsumerInfo kafkaConsumerInfo) {
   }
 
   public static String extractClusterId(
-      KafkaConsumerInfo kafkaConsumerInfo, ContextStore<Metadata, String> metadataContextStore) {
+      KafkaConsumerInfo kafkaConsumerInfo,
+      ContextStore<Metadata, MetadataState> metadataContextStore) {
     if (Config.get().isDataStreamsEnabled() && kafkaConsumerInfo != null) {
       Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
       if (consumerMetadata != null) {
-        return metadataContextStore.get(consumerMetadata);
+        MetadataState state = metadataContextStore.get(consumerMetadata);
+        return state != null ? state.clusterId : null;
       }
     }
     return null;