Fix service name overrides in consumers

Author: piochelepiotr

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DataStreamContextExtractor.java

@@ -12,14 +12,14 @@ public class DataStreamContextExtractor implements HttpCodec.Extractor {
   private final TimeSource timeSource;
   private final Supplier<TraceConfig> traceConfigSupplier;
   private final long hashOfKnownTags;
-  private final String serviceNameOverride;
+  private final ThreadLocal<String> serviceNameOverride;
 
   public DataStreamContextExtractor(
       HttpCodec.Extractor delegate,
       TimeSource timeSource,
       Supplier<TraceConfig> traceConfigSupplier,
       long hashOfKnownTags,
-      String serviceNameOverride) {
+      ThreadLocal<String> serviceNameOverride) {
     this.delegate = delegate;
     this.timeSource = timeSource;
     this.traceConfigSupplier = traceConfigSupplier;
@@ -41,7 +41,7 @@ public <C> TagContext extract(C carrier, AgentPropagation.ContextVisitor<C> gett
       if (shouldExtractPathwayContext) {
         DefaultPathwayContext pathwayContext =
             DefaultPathwayContext.extract(
-                carrier, getter, this.timeSource, this.hashOfKnownTags, serviceNameOverride);
+                carrier, getter, this.timeSource, this.hashOfKnownTags, serviceNameOverride.get());
 
         extracted.withPathwayContext(pathwayContext);
       }
@@ -50,7 +50,7 @@ public <C> TagContext extract(C carrier, AgentPropagation.ContextVisitor<C> gett
     } else if (traceConfigSupplier.get().isDataStreamsEnabled()) {
       DefaultPathwayContext pathwayContext =
           DefaultPathwayContext.extract(
-              carrier, getter, this.timeSource, this.hashOfKnownTags, serviceNameOverride);
+              carrier, getter, this.timeSource, this.hashOfKnownTags, serviceNameOverride.get());
 
       if (pathwayContext != null) {
         extracted = new TagContext();

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DataStreamPropagator.java

@@ -20,13 +20,13 @@ public class DataStreamPropagator implements Propagator {
   private final Supplier<TraceConfig> traceConfigSupplier;
   private final TimeSource timeSource;
   private final long hashOfKnownTags;
-  private final String serviceNameOverride;
+  private final ThreadLocal<String> serviceNameOverride;
 
   public DataStreamPropagator(
       Supplier<TraceConfig> traceConfigSupplier,
       TimeSource timeSource,
       long hashOfKnownTags,
-      String serviceNameOverride) {
+      ThreadLocal<String> serviceNameOverride) {
     this.traceConfigSupplier = traceConfigSupplier;
     this.timeSource = timeSource;
     this.hashOfKnownTags = hashOfKnownTags;
@@ -78,6 +78,6 @@ private boolean isDsmEnabled(@Nullable TagContext tagContext) {
 
   private <C> PathwayContext extractDsmPathwayContext(C carrier, CarrierVisitor<C> visitor) {
     return DefaultPathwayContext.extract(
-        carrier, visitor, this.timeSource, this.hashOfKnownTags, this.serviceNameOverride);
+        carrier, visitor, this.timeSource, this.hashOfKnownTags, serviceNameOverride.get());
   }
 }

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java

@@ -204,13 +204,13 @@ public PathwayContext newPathwayContext() {
   @Override
   public Propagator propagator() {
     return new DataStreamPropagator(
-        this.traceConfigSupplier, this.timeSource, this.hashOfKnownTags, getThreadServiceName());
+        this.traceConfigSupplier, this.timeSource, this.hashOfKnownTags, serviceNameOverride);
   }
 
   @Override
   public HttpCodec.Extractor extractor(HttpCodec.Extractor delegate) {
     return new DataStreamContextExtractor(
-        delegate, timeSource, traceConfigSupplier, hashOfKnownTags, getThreadServiceName());
+        delegate, timeSource, traceConfigSupplier, hashOfKnownTags, serviceNameOverride);
   }
 
   @Override