One more refactoring

kr-igor · kr-igor · commit 18258b08ec3a · 2025-07-16T21:12:40.000-05:00
diff --git a/dd-java-agent/instrumentation/armeria/armeria-grpc-0.84/src/test/groovy/ArmeriaGrpcTest.groovy b/dd-java-agent/instrumentation/armeria/armeria-grpc-0.84/src/test/groovy/ArmeriaGrpcTest.groovy
@@ -256,12 +256,12 @@ abstract class ArmeriaGrpcTest extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")
+        tags.hasAllTags("direction:in", "topic:somequeue", "type:grpc")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags("direction:in", "type:grpc")
+        tags.hasAllTags("direction:in", "type:grpc")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/dsmTest/groovy/AWS1KinesisClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/dsmTest/groovy/AWS1KinesisClientTest.groovy
@@ -116,7 +116,7 @@ abstract class AWS1KinesisClientTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:" + streamArn, "type:kinesis")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:" + streamArn, "type:kinesis")
         }
       }
       verifyAll {
diff --git a/dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/dsmTest/groovy/AWS1SnsClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/dsmTest/groovy/AWS1SnsClientTest.groovy
@@ -97,7 +97,7 @@ abstract class AWS1SnsClientTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:" + topicName, "type:sns")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:" + topicName, "type:sns")
         }
       }
       verifyAll {
diff --git a/dd-java-agent/instrumentation/aws-java-sdk-2.2/src/dsmTest/groovy/Aws2KinesisDataStreamsTest.groovy b/dd-java-agent/instrumentation/aws-java-sdk-2.2/src/dsmTest/groovy/Aws2KinesisDataStreamsTest.groovy
@@ -157,7 +157,7 @@ abstract class Aws2KinesisDataStreamsTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")
         }
       }
       verifyAll {
@@ -278,7 +278,7 @@ abstract class Aws2KinesisDataStreamsTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")
         }
       }
       verifyAll {
diff --git a/dd-java-agent/instrumentation/aws-java-sdk-2.2/src/dsmTest/groovy/Aws2SnsDataStreamsTest.groovy b/dd-java-agent/instrumentation/aws-java-sdk-2.2/src/dsmTest/groovy/Aws2SnsDataStreamsTest.groovy
@@ -145,7 +145,7 @@ abstract class Aws2SnsDataStreamsTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")
         }
       }
       verifyAll {
@@ -243,7 +243,7 @@ abstract class Aws2SnsDataStreamsTest extends VersionedNamingTestBase {
         pathwayLatencyCount += group.pathwayLatency.count
         edgeLatencyCount += group.edgeLatency.count
         verifyAll(group) {
-          tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")
+          tags.hasAllTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")
         }
       }
       verifyAll {
diff --git a/dd-java-agent/instrumentation/aws-java-sns-1.0/src/test/groovy/SnsClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sns-1.0/src/test/groovy/SnsClientTest.groovy
@@ -195,7 +195,7 @@ abstract class SnsClientTest extends VersionedNamingTestBase {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
 
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:testtopic", "type:sns")
+        tags.hasAllTags("direction:out", "topic:testtopic", "type:sns")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/aws-java-sns-2.0/src/test/groovy/SnsClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sns-2.0/src/test/groovy/SnsClientTest.groovy
@@ -165,7 +165,7 @@ abstract class SnsClientTest extends VersionedNamingTestBase {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
 
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:testtopic", "type:sns")
+        tags.hasAllTags("direction:out", "topic:testtopic", "type:sns")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/aws-java-sqs-1.0/src/test/groovy/SqsClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sqs-1.0/src/test/groovy/SqsClientTest.groovy
@@ -173,12 +173,12 @@ abstract class SqsClientTest extends VersionedNamingTestBase {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
 
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:somequeue", "type:sqs")
+        tags.hasAllTags("direction:out", "topic:somequeue", "type:sqs")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")
+        tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")
       }
     }
 
@@ -628,7 +628,7 @@ class SqsClientV1DataStreamsForkedTest extends SqsClientTest {
     StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == -2734507826469073289 }
 
     verifyAll(first) {
-      tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")
+      tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")
     }
 
     cleanup:
@@ -657,7 +657,7 @@ class SqsClientV1DataStreamsForkedTest extends SqsClientTest {
     StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
 
     verifyAll(first) {
-      tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")
+      tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")
     }
 
     cleanup:
diff --git a/dd-java-agent/instrumentation/aws-java-sqs-2.0/src/test/groovy/SqsClientTest.groovy b/dd-java-agent/instrumentation/aws-java-sqs-2.0/src/test/groovy/SqsClientTest.groovy
@@ -174,12 +174,12 @@ abstract class SqsClientTest extends VersionedNamingTestBase {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
 
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:somequeue", "type:sqs")
+        tags.hasAllTags("direction:out", "topic:somequeue", "type:sqs")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")
+        tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/google-pubsub/src/test/groovy/PubSubTest.groovy b/dd-java-agent/instrumentation/google-pubsub/src/test/groovy/PubSubTest.groovy
@@ -40,10 +40,6 @@ import spock.lang.Shared
 
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.CountDownLatch
-import java.util.function.Function
-import java.util.function.ToDoubleFunction
-import java.util.function.ToIntFunction
-import java.util.function.ToLongFunction
 
 abstract class PubSubTest extends VersionedNamingTestBase {
   private static final String PROJECT_ID = "dd-trace-java"
@@ -238,11 +234,11 @@ abstract class PubSubTest extends VersionedNamingTestBase {
 
       StatsGroup sendStat = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0}
       verifyAll (sendStat) {
-        tags == DataStreamsTags.fromTags("direction:out" , "topic:test-topic", "type:google-pubsub")
+        tags.hasAllTags("direction:out" , "topic:test-topic", "type:google-pubsub")
       }
       StatsGroup receiveStat = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == sendStat.hash}
       verifyAll(receiveStat) {
-        tags == DataStreamsTags.fromTags("direction:in" , "subscription:my-subscription", "type:google-pubsub")
+        tags.hasAllTags("direction:in" , "subscription:my-subscription", "type:google-pubsub")
         pathwayLatency.count == 1
         pathwayLatency.minValue > 0.0
         edgeLatency.count == 1
diff --git a/dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcTest.groovy b/dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcTest.groovy
@@ -245,12 +245,12 @@ abstract class GrpcTest extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "type:grpc")
+        tags.hasAllTags("direction:out", "type:grpc")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags("direction:in", "type:grpc")
+        tags.hasAllTags("direction:in", "type:grpc")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/kafka-clients-0.11/src/latestDepTest/groovy/KafkaClientTestBase.groovy b/dd-java-agent/instrumentation/kafka-clients-0.11/src/latestDepTest/groovy/KafkaClientTestBase.groovy
@@ -233,15 +233,15 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     new String(headers.headers("x-datadog-parent-id").iterator().next().value()) == "${traces[produceTraceIdx][2].spanId}"
 
     if (isDataStreamsEnabled()) {
-      def val = DataStreamsTags.fromTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
+      def val = DataStreamsTags.hasAllTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
         tags.toString() == val.toString()
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId",
diff --git a/dd-java-agent/instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy b/dd-java-agent/instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy
@@ -285,12 +285,12 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
+        tags.hasAllTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId",
@@ -431,7 +431,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:out",
           "kafka_cluster_id:$clusterId".toString(),
           "topic:$SHARED_TOPIC".toString(),
@@ -441,7 +441,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId".toString(),
@@ -919,12 +919,12 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
+        tags.hasAllTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId",
diff --git a/dd-java-agent/instrumentation/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy b/dd-java-agent/instrumentation/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy
@@ -261,11 +261,11 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
+        tags.hasAllTags("direction:out", "kafka_cluster_id:$clusterId", "topic:$SHARED_TOPIC".toString(), "type:kafka")
       }
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId",
@@ -411,7 +411,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:out",
           "kafka_cluster_id:$clusterId".toString(),
           "topic:$SHARED_TOPIC".toString(),
@@ -421,7 +421,7 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase {
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
       verifyAll(second) {
-        tags == DataStreamsTags.fromTags(
+        tags.hasAllTags(
           "direction:in",
           "group:sender",
           "kafka_cluster_id:$clusterId".toString(),
diff --git a/dd-java-agent/instrumentation/kafka-connect-0.11/src/test/groovy/ConnectWorkerInstrumentationTest.groovy b/dd-java-agent/instrumentation/kafka-connect-0.11/src/test/groovy/ConnectWorkerInstrumentationTest.groovy
@@ -154,21 +154,20 @@ class ConnectWorkerInstrumentationTest extends AgentTestRunner {
 
     StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
     verifyAll(first) {
-      tags == DataStreamsTags.fromTags(
+      tags.hasAllTags(
       "direction:out",
       "topic:test-topic",
       "type:kafka"
       )
     }
 
     StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
+    def control = DataStreamsTags.hasAllTags("direction:in", "group:test-consumer-group", "topic:test-topic", "type:kafka")
     verifyAll(second) {
-      tags == DataStreamsTags.fromTags(
-      "direction:in",
-      "group:test-consumer-group",
-      "topic:test-topic",
-      "type:kafka"
-      )
+      tags.direction == control.direction
+      tags.group == control.group
+      tags.topic == control.topic
+      tags.type == control.type
     }
     TEST_DATA_STREAMS_WRITER.getServices().contains('file-source-connector')
 
@@ -286,21 +285,17 @@ class ConnectWorkerInstrumentationTest extends AgentTestRunner {
 
     StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
     verifyAll(first) {
-      tags == DataStreamsTags.fromTags(
-      "direction:out",
-      "topic:test-topic",
-      "type:kafka"
-      )
+      tags.hasAllTags("direction:out", "topic:test-topic", "type:kafka", "kafka_cluster_id:" + clusterId)
     }
 
     StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
+    def control = DataStreamsTags.hasAllTags("direction:in", "group:connect-file-sink-connector", "topic:test-topic", "type:kafka", "kafka_cluster_id:" + clusterId)
     verifyAll(second) {
-      tags == DataStreamsTags.fromTags(
-      "direction:in",
-      "group:connect-file-sink-connector",
-      "topic:test-topic",
-      "type:kafka"
-      )
+      tags.direction == control.direction
+      tags.group == control.group
+      tags.topic == control.topic
+      tags.type == control.type
+      tags.kafkaClusterId == control.kafkaClusterId
     }
     TEST_DATA_STREAMS_WRITER.getServices().contains('file-sink-connector')
 
diff --git a/dd-java-agent/instrumentation/kafka-streams-0.11/src/latestDepTest/groovy/KafkaStreamsTest.groovy b/dd-java-agent/instrumentation/kafka-streams-0.11/src/latestDepTest/groovy/KafkaStreamsTest.groovy
@@ -227,25 +227,25 @@ class KafkaStreamsTest extends AgentTestRunner {
     if (isDataStreamsEnabled()) {
       StatsGroup originProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(originProducerPoint) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:$STREAM_PENDING", "type:kafka")
+        tags.hasAllTags("direction:out", "topic:$STREAM_PENDING", "type:kafka")
       }
 
       StatsGroup kafkaStreamsConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == originProducerPoint.hash }
       verifyAll(kafkaStreamsConsumerPoint) {
-        tags == DataStreamsTags.fromTags("direction:in",
+        tags.hasAllTags("direction:in",
           "group:test-application",
           "topic:$STREAM_PENDING".toString(),
           "type:kafka")
       }
 
       StatsGroup kafkaStreamsProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsConsumerPoint.hash }
       verifyAll(kafkaStreamsProducerPoint) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:$STREAM_PROCESSED", "type:kafka")
+        tags.hasAllTags("direction:out", "topic:$STREAM_PROCESSED", "type:kafka")
       }
 
       StatsGroup finalConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsProducerPoint.hash }
       verifyAll(finalConsumerPoint) {
-        tags == DataStreamsTags.fromTags("direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka")
+        tags.hasAllTags("direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/kafka-streams-0.11/src/test/groovy/KafkaStreamsTestBase.groovy b/dd-java-agent/instrumentation/kafka-streams-0.11/src/test/groovy/KafkaStreamsTestBase.groovy
@@ -290,25 +290,25 @@ abstract class KafkaStreamsTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup originProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(originProducerPoint) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:$STREAM_PENDING", "type:kafka")
+        tags.hasAllTags("direction:out", "topic:$STREAM_PENDING", "type:kafka")
       }
 
       StatsGroup kafkaStreamsConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == originProducerPoint.hash }
       verifyAll(kafkaStreamsConsumerPoint) {
-        tags == DataStreamsTags.fromTags("direction:in",
+        tags.hasAllTags("direction:in",
           "group:test-application",
           "topic:$STREAM_PENDING".toString(),
           "type:kafka")
       }
 
       StatsGroup kafkaStreamsProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsConsumerPoint.hash }
       verifyAll(kafkaStreamsProducerPoint) {
-        tags == DataStreamsTags.fromTags("direction:out", "topic:$STREAM_PROCESSED", "type:kafka")
+        tags.hasAllTags("direction:out", "topic:$STREAM_PROCESSED", "type:kafka")
       }
 
       StatsGroup finalConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsProducerPoint.hash }
       verifyAll(finalConsumerPoint) {
-        tags == DataStreamsTags.fromTags("direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka")
+        tags.hasAllTags("direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka")
       }
     }
 
diff --git a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy
diff --git a/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java b/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java
diff --git a/internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy

Original file line number	Diff line number	Diff line change
`@@ -256,12 +256,12 @@ abstract class ArmeriaGrpcTest extends VersionedNamingTestBase {`
`256`	`256`	`if (isDataStreamsEnabled()) {`
`257`	`257`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`258`	`258`	`verifyAll(first) {`
`259`		`- tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")`
	`259`	`+ tags.hasAllTags("direction:in", "topic:somequeue", "type:grpc")`
`260`	`260`	`}`
`261`	`261`
`262`	`262`	`StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }`
`263`	`263`	`verifyAll(second) {`
`264`		`- tags == DataStreamsTags.fromTags("direction:in", "type:grpc")`
	`264`	`+ tags.hasAllTags("direction:in", "type:grpc")`
`265`	`265`	`}`
`266`	`266`	`}`
`267`	`267`
Original file line number	Diff line number	Diff line change
`@@ -116,7 +116,7 @@ abstract class AWS1KinesisClientTest extends VersionedNamingTestBase {`
`116`	`116`	`pathwayLatencyCount += group.pathwayLatency.count`
`117`	`117`	`edgeLatencyCount += group.edgeLatency.count`
`118`	`118`	`verifyAll(group) {`
`119`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:" + streamArn, "type:kinesis")`
	`119`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:" + streamArn, "type:kinesis")`
`120`	`120`	`}`
`121`	`121`	`}`
`122`	`122`	`verifyAll {`
Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,7 @@ abstract class AWS1SnsClientTest extends VersionedNamingTestBase {`
`97`	`97`	`pathwayLatencyCount += group.pathwayLatency.count`
`98`	`98`	`edgeLatencyCount += group.edgeLatency.count`
`99`	`99`	`verifyAll(group) {`
`100`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:" + topicName, "type:sns")`
	`100`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:" + topicName, "type:sns")`
`101`	`101`	`}`
`102`	`102`	`}`
`103`	`103`	`verifyAll {`
Original file line number	Diff line number	Diff line change
`@@ -157,7 +157,7 @@ abstract class Aws2KinesisDataStreamsTest extends VersionedNamingTestBase {`
`157`	`157`	`pathwayLatencyCount += group.pathwayLatency.count`
`158`	`158`	`edgeLatencyCount += group.edgeLatency.count`
`159`	`159`	`verifyAll(group) {`
`160`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")`
	`160`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")`
`161`	`161`	`}`
`162`	`162`	`}`
`163`	`163`	`verifyAll {`
`@@ -278,7 +278,7 @@ abstract class Aws2KinesisDataStreamsTest extends VersionedNamingTestBase {`
`278`	`278`	`pathwayLatencyCount += group.pathwayLatency.count`
`279`	`279`	`edgeLatencyCount += group.edgeLatency.count`
`280`	`280`	`verifyAll(group) {`
`281`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")`
	`281`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:arnprefix:stream/somestream", "type:kinesis")`
`282`	`282`	`}`
`283`	`283`	`}`
`284`	`284`	`verifyAll {`
Original file line number	Diff line number	Diff line change
`@@ -145,7 +145,7 @@ abstract class Aws2SnsDataStreamsTest extends VersionedNamingTestBase {`
`145`	`145`	`pathwayLatencyCount += group.pathwayLatency.count`
`146`	`146`	`edgeLatencyCount += group.edgeLatency.count`
`147`	`147`	`verifyAll(group) {`
`148`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")`
	`148`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")`
`149`	`149`	`}`
`150`	`150`	`}`
`151`	`151`	`verifyAll {`
`@@ -243,7 +243,7 @@ abstract class Aws2SnsDataStreamsTest extends VersionedNamingTestBase {`
`243`	`243`	`pathwayLatencyCount += group.pathwayLatency.count`
`244`	`244`	`edgeLatencyCount += group.edgeLatency.count`
`245`	`245`	`verifyAll(group) {`
`246`		`- tags == DataStreamsTags.fromTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")`
	`246`	`+ tags.hasAllTags("direction:" + dsmDirection, "topic:mytopic", "type:sns")`
`247`	`247`	`}`
`248`	`248`	`}`
`249`	`249`	`verifyAll {`
Original file line number	Diff line number	Diff line change
`@@ -195,7 +195,7 @@ abstract class SnsClientTest extends VersionedNamingTestBase {`
`195`	`195`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`196`	`196`
`197`	`197`	`verifyAll(first) {`
`198`		`- tags == DataStreamsTags.fromTags("direction:out", "topic:testtopic", "type:sns")`
	`198`	`+ tags.hasAllTags("direction:out", "topic:testtopic", "type:sns")`
`199`	`199`	`}`
`200`	`200`	`}`
`201`	`201`
Original file line number	Diff line number	Diff line change
`@@ -165,7 +165,7 @@ abstract class SnsClientTest extends VersionedNamingTestBase {`
`165`	`165`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`166`	`166`
`167`	`167`	`verifyAll(first) {`
`168`		`- tags == DataStreamsTags.fromTags("direction:out", "topic:testtopic", "type:sns")`
	`168`	`+ tags.hasAllTags("direction:out", "topic:testtopic", "type:sns")`
`169`	`169`	`}`
`170`	`170`	`}`
`171`	`171`
Original file line number	Diff line number	Diff line change
`@@ -173,12 +173,12 @@ abstract class SqsClientTest extends VersionedNamingTestBase {`
`173`	`173`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`174`	`174`
`175`	`175`	`verifyAll(first) {`
`176`		`- tags == DataStreamsTags.fromTags("direction:out", "topic:somequeue", "type:sqs")`
	`176`	`+ tags.hasAllTags("direction:out", "topic:somequeue", "type:sqs")`
`177`	`177`	`}`
`178`	`178`
`179`	`179`	`StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }`
`180`	`180`	`verifyAll(second) {`
`181`		`- tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")`
	`181`	`+ tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")`
`182`	`182`	`}`
`183`	`183`	`}`
`184`	`184`
`@@ -628,7 +628,7 @@ class SqsClientV1DataStreamsForkedTest extends SqsClientTest {`
`628`	`628`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == -2734507826469073289 }`
`629`	`629`
`630`	`630`	`verifyAll(first) {`
`631`		`- tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")`
	`631`	`+ tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")`
`632`	`632`	`}`
`633`	`633`
`634`	`634`	`cleanup:`
`@@ -657,7 +657,7 @@ class SqsClientV1DataStreamsForkedTest extends SqsClientTest {`
`657`	`657`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`658`	`658`
`659`	`659`	`verifyAll(first) {`
`660`		`- tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")`
	`660`	`+ tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")`
`661`	`661`	`}`
`662`	`662`
`663`	`663`	`cleanup:`
Original file line number	Diff line number	Diff line change
`@@ -174,12 +174,12 @@ abstract class SqsClientTest extends VersionedNamingTestBase {`
`174`	`174`	`StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }`
`175`	`175`
`176`	`176`	`verifyAll(first) {`
`177`		`- tags == DataStreamsTags.fromTags("direction:out", "topic:somequeue", "type:sqs")`
	`177`	`+ tags.hasAllTags("direction:out", "topic:somequeue", "type:sqs")`
`178`	`178`	`}`
`179`	`179`
`180`	`180`	`StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }`
`181`	`181`	`verifyAll(second) {`
`182`		`- tags == DataStreamsTags.fromTags("direction:in", "topic:somequeue", "type:sqs")`
	`182`	`+ tags.hasAllTags("direction:in", "topic:somequeue", "type:sqs")`
`183`	`183`	`}`
`184`	`184`	`}`
`185`	`185`