Fixed even more tests

Author: kr-igor

dd-java-agent/instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy

@@ -919,8 +919,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        edgeTags == ["direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka"]
-        edgeTags.size() == 4
+        tags == DataStreamsTags.fromTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }

dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy

@@ -638,7 +638,6 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
         tags == DSM_EDGE_TAGS
-        tags.getSize() == DSM_EDGE_TAGS.getSize()
       }
     }
 
@@ -726,7 +725,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == edgeTags
+        tags == DSM_EDGE_TAGS
       }
     }
 

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

@@ -87,9 +87,6 @@ public synchronized void setCheckpoint(
     long startNanos = timeSource.getCurrentTimeNanos();
     long nanoTicks = timeSource.getNanoTicks();
 
-    PathwayHashBuilder pathwayHashBuilder =
-        new PathwayHashBuilder(hashOfKnownTags, serviceNameOverride);
-
     if (!started) {
       long defaultTimestamp = context.defaultTimestamp();
       if (defaultTimestamp == 0) {
@@ -146,7 +143,6 @@ public synchronized void setCheckpoint(
     hash = newHash;
 
     pointConsumer.accept(point);
-    log.debug("Checkpoint set {}, hash source: {}", this, pathwayHashBuilder);
   }
 
   @Override
@@ -272,21 +268,6 @@ private static DefaultPathwayContext decode(
         serviceNameOverride);
   }
 
-  private static class PathwayHashBuilder {
-    private long hash;
-
-    public PathwayHashBuilder(long baseHash, String serviceNameOverride) {
-      hash = baseHash;
-      if (serviceNameOverride != null) {
-        addTag(serviceNameOverride);
-      }
-    }
-
-    public void addTag(String tag) {
-      hash = FNV64Hash.continueHash(hash, tag, FNV64Hash.Version.v1);
-    }
-  }
-
   public static long getBaseHash(WellKnownTags wellKnownTags) {
     StringBuilder builder = new StringBuilder();
     builder.append(wellKnownTags.getService());

internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy

@@ -64,4 +64,24 @@ class DataStreamsTagsTest extends Specification {
     one != three
     two != three
   }
+
+  def 'test from tags'() {
+    setup:
+    def one = DataStreamsTags.fromTags("direction:in", "topic:abc")
+    expect:
+    one.nonNullSize() == 2
+    one.direction == "direction:in"
+    one.topic == "topic:abc"
+  }
+
+  def 'test create'() {
+    setup:
+    def one = DataStreamsTags.create("type", DataStreamsTags.Direction.Outbound)
+    def two = DataStreamsTags.create("type", DataStreamsTags.Direction.Outbound, "topic")
+    def three = DataStreamsTags.create("type", DataStreamsTags.Direction.Outbound, "topic", "group", "cluster")
+    expect:
+    one == DataStreamsTags.fromTags("type:type", "direction:out")
+    two == DataStreamsTags.fromTags("type:type", "direction:out", "topic:topic")
+    three == DataStreamsTags.fromTags("type:type", "direction:out", "topic:topic", "group:group", "kafka_cluster_id:cluster")
+  }
 }