Add base hash support and service name overrides

Author: kr-igor

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

@@ -123,6 +123,9 @@ public DefaultDataStreamsMonitoring(
 
     this.propagator =
         new DataStreamsPropagator(this, this.timeSource, this.hashOfKnownTags, serviceNameOverride);
+    // configure global tags behavior
+    DataStreamsTags.setGlobalBaseHash(this.hashOfKnownTags);
+    DataStreamsTags.setServiceNameOverride(serviceNameOverride);
   }
 
   @Override

internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java

@@ -2,6 +2,8 @@
 
 import datadog.trace.util.FNV64Hash;
 
+import javax.xml.crypto.Data;
+
 public class DataStreamsTags {
   public enum Direction {
     Unknown,
@@ -50,6 +52,9 @@ public enum Direction {
   public static final String HAS_ROUTING_KEY_TAG = "has_routing_key";
   public static final String KAFKA_CLUSTER_ID_TAG = "kafka_cluster_id";
 
+  private static volatile ThreadLocal<String> serviceNameOverride;
+  private static volatile long baseHash;
+
   public static byte[] longToBytes(long val) {
     return new byte[] {
       (byte) val,
@@ -144,6 +149,14 @@ public static DataStreamsTags createWithDataset(
         null);
   }
 
+  public static void setServiceNameOverride(ThreadLocal<String> serviceNameOverride) {
+    DataStreamsTags.serviceNameOverride = serviceNameOverride;
+  }
+
+  public static void setGlobalBaseHash(long hash) {
+    DataStreamsTags.baseHash = hash;
+  }
+
   public static DataStreamsTags createWithClusterId(
       String type, Direction direction, String topic, String clusterId) {
     return new DataStreamsTags(
@@ -209,6 +222,17 @@ public DataStreamsTags(
         kafkaClusterId != null ? KAFKA_CLUSTER_ID_TAG + ":" + kafkaClusterId : null;
     this.partition = partition != null ? PARTITION_TAG + ":" + partition : null;
 
+    if (DataStreamsTags.baseHash != 0) {
+      this.hash = DataStreamsTags.baseHash;
+    }
+
+    if (DataStreamsTags.serviceNameOverride != null) {
+      String val = DataStreamsTags.serviceNameOverride.get();
+      if (val != null) {
+        this.hash = FNV64Hash.continueHash(this.hash, val, FNV64Hash.Version.v1);
+      }
+    }
+
     // hashable tags are 0-4
     for (int i = 0; i < 7; i++) {
       String tag = this.tagByIndex(i);