diff --git a/internal-api/src/jmh/java/datadog/trace/api/datastreams/DataStreamsTagsBenchmark.java b/internal-api/src/jmh/java/datadog/trace/api/datastreams/DataStreamsTagsBenchmark.java
new file mode 100644
index 00000000000..f7435a5f9c4
--- /dev/null
+++ b/internal-api/src/jmh/java/datadog/trace/api/datastreams/DataStreamsTagsBenchmark.java
@@ -0,0 +1,66 @@
+package datadog.trace.api.datastreams;
+
+import static datadog.trace.api.datastreams.DataStreamsTags.Direction.INBOUND;
+
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Building the {@link DataStreamsTags} a Kafka consumer builds per record. With Data Streams off,
+ * nothing reads the tags ({@link #dsmOff}); with it on, the checkpoint reads their hashes ({@link
+ * #dsmOn}). Run with {@code -prof gc} for allocation per operation. The service name override is
+ * installed, as Data Streams monitoring does at startup, so building the tags reads it.
+ *
+ * Results (MacBook, JDK 21, default flags, 5 forks via {@code -Pjmh.forks=5}, 4 threads; the
+ * machine was busy, so treat the times as approximate):
+ *
+ *
+ * Benchmark eager tags (before) lazy tags (after)
+ * ns/op B/op ns/op B/op
+ * dsmOff 230.4 +- 4.4 1464 8.5 +- 0.1 112
+ * dsmOn 226.9 +- 3.4 1464 118.4 +- 1.6 112
+ *
+ */
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Fork(2)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Threads(4)
+@State(Scope.Benchmark)
+public class DataStreamsTagsBenchmark {
+ // Per-consumer values, the same instances on every record as in the instrumentation
+ final String topic = "orders";
+ final String group = "order-processor";
+ final String clusterId = "lkc-6ojk2q";
+
+ @Setup(Level.Trial)
+ public void installServiceNameOverride() {
+ DataStreamsTags.setServiceNameOverride(new ThreadLocal<>());
+ }
+
+ @Benchmark
+ public DataStreamsTags dsmOff() {
+ return DataStreamsTags.create("kafka", INBOUND, topic, group, clusterId);
+ }
+
+ @Benchmark
+ public DataStreamsTags dsmOn() {
+ DataStreamsTags tags = DataStreamsTags.create("kafka", INBOUND, topic, group, clusterId);
+ // Returned, not just its hashes, since the tags escape into the checkpoint
+ tags.getHash();
+ tags.getAggregationHash();
+ return tags;
+ }
+}
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
index df3eb214813..43435d459a9 100644
--- a/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java
+++ b/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java
@@ -13,29 +13,35 @@ public enum Direction {
public static DataStreamsTags EMPTY = DataStreamsTags.create(null, null);
+ // Hashes are computed on first read, since with Data Streams off nothing reads them
private long hash;
private long aggregationHash;
private long completeHash;
private int nonNullSize;
+ private volatile boolean hashed;
+ // Read when the tags are built, as they depend on the time and thread of the checkpoint
+ private final long baseHash;
+ private final String serviceNameOverrideValue;
+
+ // Tag values; the "name:value" tags are built from them only when read
// hash tags
- protected final String bus;
- protected final String direction;
- protected final Direction directionValue;
- protected final String exchange;
- protected final String topic;
- protected final String type;
- protected final String subscription;
+ private final String bus;
+ private final Direction directionValue;
+ private final String exchange;
+ private final String topic;
+ private final String type;
+ private final String subscription;
// additional grouping tags
- protected final String datasetName;
- protected final String datasetNamespace;
- protected final String isManual;
+ private final String datasetName;
+ private final String datasetNamespace;
+ private final Boolean isManual;
// informational tags
- protected final String group;
- protected final String consumerGroup;
- protected final String hasRoutingKey;
- protected final String kafkaClusterId;
- protected final String partition;
+ private final String group;
+ private final String consumerGroup;
+ private final Boolean hasRoutingKey;
+ private final String kafkaClusterId;
+ private final String partition;
public static final String MANUAL_TAG = "manual_checkpoint";
public static final String TYPE_TAG = "type";
@@ -145,92 +151,28 @@ public static DataStreamsTags createWithPartition(
/// For usage in tests *only*
public Boolean hasAllTags(String[] tags) {
for (String tag : tags) {
- if (tag.indexOf(':') == -1) {
+ int separator = tag.indexOf(':');
+ if (separator == -1) {
return false;
}
- String key = tag.substring(0, tag.indexOf(':'));
- String value = tag.substring(tag.indexOf(':') + 1);
- switch (key) {
- case BUS_TAG:
- if (!Objects.equals(this.bus, tag)) {
- return false;
- }
- break;
- case DIRECTION_TAG:
- if (!Objects.equals(
- this.directionValue,
- Objects.equals(value, "out") ? Direction.OUTBOUND : Direction.INBOUND)) {
- return false;
- }
- break;
- case EXCHANGE_TAG:
- if (!Objects.equals(this.exchange, tag)) {
- return false;
- }
- break;
- case TOPIC_TAG:
- if (!Objects.equals(this.topic, tag)) {
- return false;
- }
- break;
- case TYPE_TAG:
- if (!Objects.equals(this.type, tag)) {
- return false;
- }
- break;
- case SUBSCRIPTION_TAG:
- if (!Objects.equals(this.subscription, tag)) {
- return false;
- }
- break;
- case DATASET_NAME_TAG:
- if (!Objects.equals(this.datasetName, tag)) {
- return false;
- }
- break;
- case DATASET_NAMESPACE_TAG:
- if (!Objects.equals(this.datasetNamespace, tag)) {
- return false;
- }
- break;
- case MANUAL_TAG:
- if (!Objects.equals(this.isManual, tag)) {
- return false;
- }
- break;
- case GROUP_TAG:
- if (!Objects.equals(this.group, tag)) {
- return false;
- }
- break;
- case CONSUMER_GROUP_TAG:
- if (!Objects.equals(this.consumerGroup, tag)) {
- return false;
- }
- break;
- case HAS_ROUTING_KEY_TAG:
- if (!Objects.equals(this.hasRoutingKey, tag)) {
- return false;
- }
- break;
- case KAFKA_CLUSTER_ID_TAG:
- if (!Objects.equals(this.kafkaClusterId, tag)) {
- return false;
- }
- break;
- case PARTITION_TAG:
- if (!Objects.equals(this.partition, tag)) {
- return false;
- }
- break;
- default:
- return false;
+ int index = indexOfName(tag.substring(0, separator));
+ if (index == -1 || !Objects.equals(tagByIndex(index), tag)) {
+ return false;
}
}
return true;
}
+ private int indexOfName(String name) {
+ for (int i = 0; i < size(); i++) {
+ if (nameByIndex(i).equals(name)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
public static DataStreamsTags createWithGroup(
String type, Direction direction, String topic, String group) {
return new DataStreamsTags(
@@ -301,68 +243,78 @@ public DataStreamsTags(
Boolean hasRoutingKey,
String kafkaClusterId,
String partition) {
- this.bus = bus != null ? BUS_TAG + ":" + bus : null;
+ this.bus = bus;
this.directionValue = direction;
- if (direction == Direction.INBOUND) {
- this.direction = DIRECTION_TAG + ":in";
- } else if (direction == Direction.OUTBOUND) {
- this.direction = DIRECTION_TAG + ":out";
- } else {
- this.direction = null;
+ this.exchange = exchange;
+ this.topic = topic;
+ this.type = type;
+ this.subscription = subscription;
+ this.datasetName = datasetName;
+ this.datasetNamespace = datasetNamespace;
+ this.isManual = isManual;
+ this.group = group;
+ this.consumerGroup = consumerGroup;
+ this.hasRoutingKey = hasRoutingKey;
+ this.kafkaClusterId = kafkaClusterId;
+ this.partition = partition;
+
+ this.baseHash = BaseHash.getBaseHash();
+ ThreadLocal serviceNameOverride = DataStreamsTags.serviceNameOverride;
+ this.serviceNameOverrideValue = serviceNameOverride != null ? serviceNameOverride.get() : null;
+ }
+
+ private void ensureHashed() {
+ if (hashed) {
+ return;
}
- this.exchange = exchange != null ? EXCHANGE_TAG + ":" + exchange : null;
- this.topic = topic != null ? TOPIC_TAG + ":" + topic : null;
- this.type = type != null ? TYPE_TAG + ":" + type : null;
- this.subscription = subscription != null ? SUBSCRIPTION_TAG + ":" + subscription : null;
- this.datasetName = datasetName != null ? DATASET_NAME_TAG + ":" + datasetName : null;
- this.datasetNamespace =
- datasetNamespace != null ? DATASET_NAMESPACE_TAG + ":" + datasetNamespace : null;
- this.isManual = isManual != null ? MANUAL_TAG + ":" + isManual : null;
- this.group = group != null ? GROUP_TAG + ":" + group : null;
- this.consumerGroup = consumerGroup != null ? CONSUMER_GROUP_TAG + ":" + consumerGroup : null;
- this.hasRoutingKey = hasRoutingKey != null ? HAS_ROUTING_KEY_TAG + ":" + hasRoutingKey : null;
- this.kafkaClusterId =
- kafkaClusterId != null ? KAFKA_CLUSTER_ID_TAG + ":" + kafkaClusterId : null;
- this.partition = partition != null ? PARTITION_TAG + ":" + partition : null;
-
- this.hash = BaseHash.getBaseHash();
-
- if (DataStreamsTags.serviceNameOverride != null) {
- String val = DataStreamsTags.serviceNameOverride.get();
- if (val != null) {
- this.hash = FNV64Hash.continueHash(this.hash, val, FNV64Hash.Version.v1);
- }
+ // Racing threads compute the same values, so the race is benign
+ long hash = baseHash;
+ if (serviceNameOverrideValue != null) {
+ hash = FNV64Hash.continueHash(hash, serviceNameOverrideValue, FNV64Hash.Version.v1);
}
+ int nonNullSize = 0;
// hashable tags are 0-6: bus, direction, exchange, topic, type, subscription, kafkaClusterId
for (int i = 0; i < 7; i++) {
- String tag = this.tagByIndex(i);
- if (tag != null) {
- this.nonNullSize++;
- this.hash = FNV64Hash.continueHash(this.hash, tag, FNV64Hash.Version.v1);
+ String value = valueByIndex(i);
+ if (value != null) {
+ nonNullSize++;
+ hash = continueHash(hash, i, value);
}
}
// aggregation tags are 7-11: datasetName, datasetNamespace, isManual, group, consumerGroup
- this.aggregationHash = this.hash;
+ long aggregationHash = hash;
for (int i = 7; i < 12; i++) {
- String tag = this.tagByIndex(i);
- if (tag != null) {
- this.nonNullSize++;
- this.aggregationHash =
- FNV64Hash.continueHash(this.aggregationHash, tag, FNV64Hash.Version.v1);
+ String value = valueByIndex(i);
+ if (value != null) {
+ nonNullSize++;
+ aggregationHash = continueHash(aggregationHash, i, value);
}
}
// values are 12-13: partition, hasRoutingKey
- this.completeHash = aggregationHash;
- for (int i = 12; i < this.size(); i++) {
- String tag = this.tagByIndex(i);
- if (tag != null) {
- this.nonNullSize++;
- this.completeHash = FNV64Hash.continueHash(this.completeHash, tag, FNV64Hash.Version.v1);
+ long completeHash = aggregationHash;
+ for (int i = 12; i < size(); i++) {
+ String value = valueByIndex(i);
+ if (value != null) {
+ nonNullSize++;
+ completeHash = continueHash(completeHash, i, value);
}
}
+
+ this.hash = hash;
+ this.aggregationHash = aggregationHash;
+ this.completeHash = completeHash;
+ this.nonNullSize = nonNullSize;
+ this.hashed = true;
+ }
+
+ /** Continues {@code hash} over the tag {@code "name:value"}, without building it. */
+ private static long continueHash(long hash, int index, String value) {
+ hash = FNV64Hash.continueHash(hash, nameByIndex(index), FNV64Hash.Version.v1);
+ hash = FNV64Hash.continueHash(hash, ":", FNV64Hash.Version.v1);
+ return FNV64Hash.continueHash(hash, value, FNV64Hash.Version.v1);
}
public int size() {
@@ -372,12 +324,22 @@ public int size() {
// WARNING: DO NOT REORDER! Indices 0-6 are hash tags, 7-11 are aggregation tags, 12-13 are
// values. Reordering breaks hashing!
+ /**
+ * The {@code name:value} tag at {@code index}, or null if it has no value. Builds a new string on
+ * each call, as do the {@code get*} tag getters, so they're meant for serialization and tests,
+ * not per-message use.
+ */
public String tagByIndex(int index) {
+ String value = valueByIndex(index);
+ return value != null ? nameByIndex(index) + ":" + value : null;
+ }
+
+ private String valueByIndex(int index) {
switch (index) {
case 0:
return this.bus;
case 1:
- return this.direction;
+ return directionTagValue();
case 2:
return this.exchange;
case 3:
@@ -393,7 +355,7 @@ public String tagByIndex(int index) {
case 8:
return this.datasetNamespace;
case 9:
- return this.isManual;
+ return toString(this.isManual);
case 10:
return this.group;
case 11:
@@ -401,34 +363,82 @@ public String tagByIndex(int index) {
case 12:
return this.partition;
case 13:
- return this.hasRoutingKey;
+ return toString(this.hasRoutingKey);
default:
return null;
}
}
+ private static String nameByIndex(int index) {
+ switch (index) {
+ case 0:
+ return BUS_TAG;
+ case 1:
+ return DIRECTION_TAG;
+ case 2:
+ return EXCHANGE_TAG;
+ case 3:
+ return TOPIC_TAG;
+ case 4:
+ return TYPE_TAG;
+ case 5:
+ return SUBSCRIPTION_TAG;
+ case 6:
+ return KAFKA_CLUSTER_ID_TAG;
+ case 7:
+ return DATASET_NAME_TAG;
+ case 8:
+ return DATASET_NAMESPACE_TAG;
+ case 9:
+ return MANUAL_TAG;
+ case 10:
+ return GROUP_TAG;
+ case 11:
+ return CONSUMER_GROUP_TAG;
+ case 12:
+ return PARTITION_TAG;
+ case 13:
+ return HAS_ROUTING_KEY_TAG;
+ default:
+ return null;
+ }
+ }
+
+ private String directionTagValue() {
+ if (directionValue == Direction.INBOUND) {
+ return "in";
+ } else if (directionValue == Direction.OUTBOUND) {
+ return "out";
+ }
+ return null;
+ }
+
+ private static String toString(Boolean value) {
+ return value != null ? value.toString() : null;
+ }
+
public String getDirection() {
- return this.direction;
+ return tagByIndex(1);
}
public String getTopic() {
- return this.topic;
+ return tagByIndex(3);
}
public String getType() {
- return this.type;
+ return tagByIndex(4);
}
public String getIsManual() {
- return this.isManual;
+ return tagByIndex(9);
}
public String getBus() {
- return this.bus;
+ return tagByIndex(0);
}
public String getExchange() {
- return this.exchange;
+ return tagByIndex(2);
}
public Direction getDirectionValue() {
@@ -436,46 +446,49 @@ public Direction getDirectionValue() {
}
public String getSubscription() {
- return this.subscription;
+ return tagByIndex(5);
}
public String getDatasetName() {
- return this.datasetName;
+ return tagByIndex(7);
}
public String getDatasetNamespace() {
- return this.datasetNamespace;
+ return tagByIndex(8);
}
public String getGroup() {
- return this.group;
+ return tagByIndex(10);
}
public String getConsumerGroup() {
- return this.consumerGroup;
+ return tagByIndex(11);
}
public String getPartition() {
- return this.partition;
+ return tagByIndex(12);
}
public String getKafkaClusterId() {
- return this.kafkaClusterId;
+ return tagByIndex(6);
}
public String getHasRoutingKey() {
- return this.hasRoutingKey;
+ return tagByIndex(13);
}
public int nonNullSize() {
+ ensureHashed();
return this.nonNullSize;
}
public long getHash() {
+ ensureHashed();
return hash;
}
public long getAggregationHash() {
+ ensureHashed();
return aggregationHash;
}
@@ -485,11 +498,14 @@ public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) return false;
DataStreamsTags that = (DataStreamsTags) o;
+ ensureHashed();
+ that.ensureHashed();
return this.completeHash == that.completeHash;
}
@Override
public int hashCode() {
+ ensureHashed();
return Long.hashCode(this.completeHash);
}
@@ -497,37 +513,37 @@ public int hashCode() {
public String toString() {
return "DataStreamsTags{"
+ "bus='"
- + this.bus
+ + getBus()
+ "', direction='"
- + this.direction
+ + getDirection()
+ "', exchange='"
- + this.exchange
+ + getExchange()
+ "', topic='"
- + this.topic
+ + getTopic()
+ "', type='"
- + this.type
+ + getType()
+ "', subscription='"
- + this.subscription
+ + getSubscription()
+ "', datasetName='"
- + this.datasetName
+ + getDatasetName()
+ "', datasetNamespace='"
- + this.datasetNamespace
+ + getDatasetNamespace()
+ "', isManual="
- + this.isManual
+ + getIsManual()
+ "', group='"
- + this.group
+ + getGroup()
+ "', consumerGroup='"
- + this.consumerGroup
+ + getConsumerGroup()
+ "', hasRoutingKey='"
- + this.hasRoutingKey
+ + getHasRoutingKey()
+ "', kafkaClusterId='"
- + this.kafkaClusterId
+ + getKafkaClusterId()
+ "', partition='"
- + this.partition
+ + getPartition()
+ "', hash='"
- + hash
+ + getHash()
+ "', aggregationHash='"
- + aggregationHash
+ + getAggregationHash()
+ "', size='"
+ size();
}
diff --git a/internal-api/src/main/java/datadog/trace/util/FNV64Hash.java b/internal-api/src/main/java/datadog/trace/util/FNV64Hash.java
index ec0120b66db..f90d181a6a6 100644
--- a/internal-api/src/main/java/datadog/trace/util/FNV64Hash.java
+++ b/internal-api/src/main/java/datadog/trace/util/FNV64Hash.java
@@ -1,7 +1,5 @@
package datadog.trace.util;
-import java.nio.charset.StandardCharsets;
-
/**
* Calculates the FNV 64 bit hash. Longs should be treated as though they were unsigned
*
@@ -17,11 +15,41 @@ public enum Version {
}
public static long generateHash(String data, Version version) {
- return generateHash(data.getBytes(StandardCharsets.UTF_8), version);
+ return continueHash(FNV_INIT, data, version);
}
+ /**
+ * Continues the hash over the UTF-8 bytes of {@code data}, as {@link String#getBytes} would
+ * encode them (an unpaired surrogate becomes {@code '?'}), without allocating them.
+ */
public static long continueHash(long currentHash, String data, Version version) {
- return continueHash(currentHash, data.getBytes(StandardCharsets.UTF_8), version);
+ boolean v1 = version == Version.v1;
+ long hash = currentHash;
+ int length = data.length();
+ for (int i = 0; i < length; i++) {
+ char c = data.charAt(i);
+ if (c < 0x80) {
+ hash = hashByte(hash, c, v1);
+ } else if (c < 0x800) {
+ hash = hashByte(hash, 0xc0 | (c >> 6), v1);
+ hash = hashByte(hash, 0x80 | (c & 0x3f), v1);
+ } else if (!Character.isSurrogate(c)) {
+ hash = hashByte(hash, 0xe0 | (c >> 12), v1);
+ hash = hashByte(hash, 0x80 | ((c >> 6) & 0x3f), v1);
+ hash = hashByte(hash, 0x80 | (c & 0x3f), v1);
+ } else if (Character.isHighSurrogate(c)
+ && i + 1 < length
+ && Character.isLowSurrogate(data.charAt(i + 1))) {
+ int codePoint = Character.toCodePoint(c, data.charAt(++i));
+ hash = hashByte(hash, 0xf0 | (codePoint >> 18), v1);
+ hash = hashByte(hash, 0x80 | ((codePoint >> 12) & 0x3f), v1);
+ hash = hashByte(hash, 0x80 | ((codePoint >> 6) & 0x3f), v1);
+ hash = hashByte(hash, 0x80 | (codePoint & 0x3f), v1);
+ } else {
+ hash = hashByte(hash, '?', v1);
+ }
+ }
+ return hash;
}
public static long generateHash(byte[] data, Version version) {
@@ -45,6 +73,10 @@ public static long continueHash(
}
}
+ private static long hashByte(long hash, int b, boolean v1) {
+ return v1 ? (hash * FNV_PRIME) ^ b : (hash ^ b) * FNV_PRIME;
+ }
+
private static long generateFNV1Hash(long currentHash, byte[] data, int start, int length) {
long hash = currentHash;
diff --git a/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsTest.java b/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsTest.java
index 9bebbf5d7b6..89232d8659c 100644
--- a/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsTest.java
+++ b/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsTest.java
@@ -1,6 +1,13 @@
package datadog.trace.api.datastreams;
+import static datadog.trace.api.BaseHash.getBaseHash;
+import static datadog.trace.api.BaseHash.updateBaseHash;
+import static datadog.trace.api.datastreams.DataStreamsTags.Direction.INBOUND;
import static datadog.trace.api.datastreams.DataStreamsTags.Direction.OUTBOUND;
+import static datadog.trace.api.datastreams.DataStreamsTags.setServiceNameOverride;
+import static datadog.trace.util.FNV64Hash.Version.v1;
+import static datadog.trace.util.FNV64Hash.continueHash;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -9,12 +16,26 @@
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import datadog.trace.api.BaseHash;
import datadog.trace.api.datastreams.DataStreamsTags.Direction;
import java.nio.ByteBuffer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class DataStreamsTagsTest {
+ private long originalBaseHash;
+
+ @BeforeEach
+ void captureBaseHash() {
+ originalBaseHash = getBaseHash();
+ }
+
+ /** The override and base hash are global, so a test changing them mustn't leak into others. */
+ @AfterEach
+ void restoreGlobals() {
+ setServiceNameOverride(null);
+ updateBaseHash(originalBaseHash);
+ }
private static DataStreamsTags getTags(int idx) {
return new DataStreamsTags(
@@ -98,6 +119,19 @@ void testHasAllTags() {
assertFalse(tags.hasAllTags(new String[] {"garbage"}));
}
+ @Test
+ void hasAllTagsRejectsTagsItDoesNotHave() {
+ DataStreamsTags tags = DataStreamsTags.create("kafka", INBOUND, "orders", "group", "cluster");
+
+ assertTrue(tags.hasAllTags(new String[] {"topic:orders", "direction:in", "group:group"}));
+ assertFalse(tags.hasAllTags(new String[] {"topic:payments"}), "different value");
+ assertFalse(tags.hasAllTags(new String[] {"exchange:orders"}), "absent tag");
+ assertFalse(tags.hasAllTags(new String[] {"unknown:orders"}), "unknown name");
+ assertFalse(tags.hasAllTags(new String[] {"direction:out"}), "different direction");
+ assertFalse(tags.hasAllTags(new String[] {"direction:garbage"}), "unknown direction");
+ assertFalse(tags.hasAllTags(new String[] {"topic:orders", "group:other"}), "one mismatch");
+ }
+
@Test
void testLongToBytes() {
long value = 123444L;
@@ -115,10 +149,10 @@ void testServiceNameOverrideAndGlobalHash() {
ThreadLocal serviceName = new ThreadLocal<>();
serviceName.set("test");
- DataStreamsTags.setServiceNameOverride(serviceName);
+ setServiceNameOverride(serviceName);
DataStreamsTags two = getTags(0);
- BaseHash.updateBaseHash(12);
+ updateBaseHash(12);
DataStreamsTags three = getTags(0);
assertNotEquals(one.getHash(), two.getHash());
@@ -489,4 +523,90 @@ void testAllThreeHashLevelsAreDifferentWhenAppropriateTagsChange() {
withPartition.getAggregationHash()); // aggregation hash unchanged
assertNotEquals(base, withPartition); // but complete hash changed
}
+
+ @Test
+ void hashesMatchHashingTheEncodedTags() {
+ DataStreamsTags[] tagsList = {
+ getTags(0),
+ DataStreamsTags.create(null, null),
+ DataStreamsTags.create("kafka", INBOUND, "orders", "group", "cluster"),
+ DataStreamsTags.createWithExchange("rabbitmq", OUTBOUND, "exchange", false),
+ DataStreamsTags.createWithDataset("s3", Direction.UNKNOWN, null, "caf\u00e9", "\u6ce8\ud83d"),
+ new DataStreamsTags(
+ "bus\ud83d\ude80",
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ "consumer",
+ null,
+ null,
+ "3")
+ };
+ for (DataStreamsTags tags : tagsList) {
+ assertHashesMatchEncodedTags(tags, getBaseHash(), null);
+ }
+ }
+
+ @Test
+ void hashesUseTheBaseHashAndServiceNameOverrideOfConstruction() {
+ long baseHash = getBaseHash();
+ ThreadLocal serviceName = new ThreadLocal<>();
+ serviceName.set("service-at-construction");
+ setServiceNameOverride(serviceName);
+ DataStreamsTags tags = getTags(1);
+
+ // Changed before the hashes are first read, which mustn't change them
+ serviceName.set("service-later");
+ updateBaseHash(baseHash + 1);
+
+ assertHashesMatchEncodedTags(tags, baseHash, "service-at-construction");
+ }
+
+ /** Checks the hashes against FNV-64 over the UTF-8 encoded "name:value" tags. */
+ private static void assertHashesMatchEncodedTags(
+ DataStreamsTags tags, long baseHash, String serviceName) {
+ long hash = baseHash;
+ if (serviceName != null) {
+ hash = hashEncoded(hash, serviceName);
+ }
+ int nonNullSize = 0;
+ for (int i = 0; i < 7; i++) {
+ String tag = tags.tagByIndex(i);
+ if (tag != null) {
+ nonNullSize++;
+ hash = hashEncoded(hash, tag);
+ }
+ }
+ long aggregationHash = hash;
+ for (int i = 7; i < 12; i++) {
+ String tag = tags.tagByIndex(i);
+ if (tag != null) {
+ nonNullSize++;
+ aggregationHash = hashEncoded(aggregationHash, tag);
+ }
+ }
+ long completeHash = aggregationHash;
+ for (int i = 12; i < tags.size(); i++) {
+ String tag = tags.tagByIndex(i);
+ if (tag != null) {
+ nonNullSize++;
+ completeHash = hashEncoded(completeHash, tag);
+ }
+ }
+ assertEquals(hash, tags.getHash(), tags::toString);
+ assertEquals(aggregationHash, tags.getAggregationHash(), tags::toString);
+ // The complete hash isn't exposed, but equals and hashCode are defined by it
+ assertEquals(Long.hashCode(completeHash), tags.hashCode(), tags::toString);
+ assertEquals(nonNullSize, tags.nonNullSize(), tags::toString);
+ }
+
+ private static long hashEncoded(long hash, String value) {
+ return continueHash(hash, value.getBytes(UTF_8), v1);
+ }
}
diff --git a/internal-api/src/test/java/datadog/trace/util/FNV64HashTest.java b/internal-api/src/test/java/datadog/trace/util/FNV64HashTest.java
index e534a87a944..301679f65ca 100644
--- a/internal-api/src/test/java/datadog/trace/util/FNV64HashTest.java
+++ b/internal-api/src/test/java/datadog/trace/util/FNV64HashTest.java
@@ -6,6 +6,7 @@
import static org.junit.jupiter.params.provider.Arguments.arguments;
import datadog.trace.util.FNV64Hash.Version;
+import java.nio.charset.StandardCharsets;
import java.util.stream.Stream;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -234,6 +235,35 @@ private static Stream binaryTestCasesArguments() {
arguments(repeat(bytes(0x7f), 500), "e6be57375ad89b99", "39e9f18f2f85e221"));
}
+ @ParameterizedTest(name = "{index}: {0}")
+ @MethodSource("utf8Strings")
+ void stringHashMatchesItsUtf8Bytes(String scenario, String input) {
+ byte[] utf8 = input.getBytes(StandardCharsets.UTF_8);
+ for (Version version : Version.values()) {
+ assertEquals(generateHash(utf8, version), generateHash(input, version), scenario);
+ assertEquals(
+ continueHash(0x1234_5678_9abc_def0L, utf8, version),
+ continueHash(0x1234_5678_9abc_def0L, input, version),
+ scenario);
+ }
+ }
+
+ // A method source, not a table: the inputs are escapes for unpaired surrogates and other
+ // characters a table can't show readably
+ static Stream utf8Strings() {
+ return Stream.of(
+ arguments("ascii", "topic:orders"),
+ arguments("two-byte", "topic:caf\u00e9"),
+ arguments("three-byte", "topic:\u6ce8\u6587"),
+ arguments("max three-byte", "\uffff"),
+ arguments("supplementary", "topic:\ud83d\ude80"),
+ arguments("lone high surrogate", "a\ud83db"),
+ arguments("high surrogate at end", "a\ud83d"),
+ arguments("lone low surrogate", "a\ude80b"),
+ arguments("reversed pair", "\ude80\ud83d"),
+ arguments("high surrogate pair then lone", "\ud83d\ude80\ud83d"));
+ }
+
private static String repeat(String value, int times) {
StringBuilder builder = new StringBuilder();
for (int i = 0; i < times; i++) {