feat(dsm): Migrate DSM injection call to propagator API 2/3

Author: PerfectSlayer

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

@@ -1,5 +1,7 @@
 package datadog.trace.instrumentation.kafka_clients;
 
+import static datadog.trace.api.datastreams.DataStreamsContext.fromKafka;
+import static datadog.trace.bootstrap.instrumentation.api.AgentPropagation.DSM_CONCERN;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateNext;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.closePrevious;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -19,7 +21,10 @@
 import static datadog.trace.instrumentation.kafka_common.Utils.computePayloadSizeBytes;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import datadog.context.propagation.Propagator;
+import datadog.context.propagation.Propagators;
 import datadog.trace.api.Config;
+import datadog.trace.api.datastreams.DataStreamsContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpanContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
@@ -116,9 +121,9 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
               // since the data received from the source may leave the topology on
               // some other instance of the application, breaking the context propagation
               // for DSM users
-              propagate()
-                  .injectPathwayContext(
-                      span, val.headers(), SETTER, sortedTags, val.timestamp(), payloadSize);
+              Propagator dsmPropagator = Propagators.forConcern(DSM_CONCERN);
+              DataStreamsContext dsmContext = fromKafka(sortedTags, val.timestamp(), payloadSize);
+              dsmPropagator.inject(span.with(dsmContext), val.headers(), SETTER);
             }
           }
         } else {

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

@@ -1,5 +1,7 @@
 package datadog.trace.instrumentation.kafka_clients38;
 
+import static datadog.trace.api.datastreams.DataStreamsContext.fromKafka;
+import static datadog.trace.bootstrap.instrumentation.api.AgentPropagation.DSM_CONCERN;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateNext;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.closePrevious;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -14,7 +16,10 @@
 import static datadog.trace.instrumentation.kafka_clients38.TextMapInjectAdapter.SETTER;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import datadog.context.propagation.Propagator;
+import datadog.context.propagation.Propagators;
 import datadog.trace.api.Config;
+import datadog.trace.api.datastreams.DataStreamsContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpanContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
@@ -116,9 +121,9 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
               // since the data received from the source may leave the topology on
               // some other instance of the application, breaking the context propagation
               // for DSM users
-              propagate()
-                  .injectPathwayContext(
-                      span, val.headers(), SETTER, sortedTags, val.timestamp(), payloadSize);
+              Propagator dsmPropagator = Propagators.forConcern(DSM_CONCERN);
+              DataStreamsContext dsmContext = fromKafka(sortedTags, val.timestamp(), payloadSize);
+              dsmPropagator.inject(span.with(dsmContext), val.headers(), SETTER);
             }
           }
         } else {

dd-java-agent/instrumentation/kafka-streams-0.11/src/main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamTaskInstrumentation.java

@@ -1,6 +1,8 @@
 package datadog.trace.instrumentation.kafka_streams;
 
 import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static datadog.trace.api.datastreams.DataStreamsContext.fromKafka;
+import static datadog.trace.bootstrap.instrumentation.api.AgentPropagation.DSM_CONCERN;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan;
@@ -29,9 +31,12 @@
 import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
 
 import com.google.auto.service.AutoService;
+import datadog.context.propagation.Propagator;
+import datadog.context.propagation.Propagators;
 import datadog.trace.agent.tooling.Instrumenter;
 import datadog.trace.agent.tooling.InstrumenterModule;
 import datadog.trace.api.Config;
+import datadog.trace.api.datastreams.DataStreamsContext;
 import datadog.trace.bootstrap.InstrumentationContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentScope;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
@@ -263,9 +268,9 @@ public static void start(
               .setCheckpoint(span, sortedTags, record.timestamp, payloadSize);
         } else {
           if (STREAMING_CONTEXT.isSourceTopic(record.topic())) {
-            propagate()
-                .injectPathwayContext(
-                    span, record, SR_SETTER, sortedTags, record.timestamp, payloadSize);
+            Propagator dsmPropagator = Propagators.forConcern(DSM_CONCERN);
+            DataStreamsContext dsmContext = fromKafka(sortedTags, record.timestamp, payloadSize);
+            dsmPropagator.inject(span.with(dsmContext), record, SR_SETTER);
           }
         }
       } else {
@@ -345,9 +350,9 @@ public static void start(
               .setCheckpoint(span, sortedTags, record.timestamp(), payloadSize);
         } else {
           if (STREAMING_CONTEXT.isSourceTopic(record.topic())) {
-            propagate()
-                .injectPathwayContext(
-                    span, record, PR_SETTER, sortedTags, record.timestamp(), payloadSize);
+            Propagator dsmPropagator = Propagators.forConcern(DSM_CONCERN);
+            DataStreamsContext dsmContext = fromKafka(sortedTags, record.timestamp(), payloadSize);
+            dsmPropagator.inject(span.with(dsmContext), record, PR_SETTER);
           }
         }
       } else {