dougqh · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026
@@ -1,7 +1,5 @@
 package datadog.trace.bootstrap.instrumentation.decorator;
 
-import static datadog.trace.bootstrap.instrumentation.java.net.HostNameResolver.hostName;
-
 import datadog.appsec.api.blocking.BlockingException;
 import datadog.context.Context;
 import datadog.context.ContextScope;
@@ -15,8 +13,6 @@
 import datadog.trace.bootstrap.instrumentation.api.ErrorPriorities;
 import datadog.trace.bootstrap.instrumentation.api.SpanPrototype;
 import java.lang.reflect.Method;
-import java.net.Inet4Address;
-import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.concurrent.ExecutionException;
@@ -194,11 +190,14 @@ protected void doOnError(final AgentSpan span, final Throwable throwable, byte e
         throwable instanceof ExecutionException ? throwable.getCause() : throwable, errorPriority);
   }
 
+  // The peer extractors write straight to the span rather than through setTagsFrom: resolving the
+  // hostname can mean a reverse DNS lookup, which mustn't run under the span's tag lock. To skip
+  // the lookup per span, freeze the extraction once per address instead (FrozenTags).
+
   public final void onPeerConnection(
       final AgentSpan span, @Nullable final InetSocketAddress remoteConnection) {
     if (remoteConnection != null) {
-      onPeerConnection(span, remoteConnection.getAddress(), !remoteConnection.isUnresolved());
-      setPeerPort(span, remoteConnection.getPort());
+      PeerSocketAddress.INSTANCE.extract(remoteConnection, span);
     }
   }
 
@@ -210,15 +209,7 @@ public final void onPeerConnection(
   public final void onPeerConnection(
       AgentSpan span, @Nullable InetAddress remoteAddress, boolean resolved) {
     if (remoteAddress != null) {
-      String ip = remoteAddress.getHostAddress();
-      if (resolved && Config.get().isPeerHostNameEnabled()) {
-        span.setTag(KnownTags.PEER_HOSTNAME_ID, hostName(remoteAddress, ip));
-      }
-      if (remoteAddress instanceof Inet4Address) {
-        span.setTag(KnownTags.PEER_IPV4_ID, ip);
-      } else if (remoteAddress instanceof Inet6Address) {
-        span.setTag(KnownTags.PEER_IPV6_ID, ip);
-      }
+      (resolved ? PeerAddress.RESOLVED : PeerAddress.UNRESOLVED).extract(remoteAddress, span);
     }
   }
 

@@ -0,0 +1,47 @@
+package datadog.trace.bootstrap.instrumentation.decorator;
+
+import static datadog.trace.bootstrap.instrumentation.java.net.HostNameResolver.hostName;
+
+import datadog.trace.api.Config;
+import datadog.trace.api.KnownTags;
+import datadog.trace.bootstrap.instrumentation.api.SpanData;
+import datadog.trace.bootstrap.instrumentation.api.TagExtractor;
+import java.net.Inet4Address;
+import java.net.Inet6Address;
+import java.net.InetAddress;
+
+/**
+ * Extracts the peer tags of a remote {@link InetAddress}: its IP, as {@code peer.ipv4} or {@code
+ * peer.ipv6}, and, when the address was resolved and peer hostnames are enabled, {@code
+ * peer.hostname}.
+ *
+ * <p>Resolving the hostname may do a reverse DNS lookup, so don't apply this with {@code
+ * span.setTagsFrom(address, extractor)}, which holds the span's tag lock while it runs: call {@code
+ * extract} on the span directly, or freeze it once per address ({@code FrozenTags.of}).
+ */
+public final class PeerAddress implements TagExtractor<InetAddress> {
+  /** For an address that was resolved from a hostname. */
+  public static final PeerAddress RESOLVED = new PeerAddress(true);
+
+  /** For an address that wasn't resolved: no {@code peer.hostname}. */
+  public static final PeerAddress UNRESOLVED = new PeerAddress(false);
+
+  private final boolean resolved;
+
+  private PeerAddress(boolean resolved) {
+    this.resolved = resolved;
+  }
+
+  @Override
+  public void extract(InetAddress address, SpanData span) {
+    String ip = address.getHostAddress();
+    if (resolved && Config.get().isPeerHostNameEnabled()) {
+      span.setTag(KnownTags.PEER_HOSTNAME_ID, hostName(address, ip));
+    }
+    if (address instanceof Inet4Address) {
+      span.setTag(KnownTags.PEER_IPV4_ID, ip);
+    } else if (address instanceof Inet6Address) {
+      span.setTag(KnownTags.PEER_IPV6_ID, ip);
+    }
+  }
+}
@@ -0,0 +1,30 @@
+package datadog.trace.bootstrap.instrumentation.decorator;
+
+import datadog.trace.api.KnownTags;
+import datadog.trace.bootstrap.instrumentation.api.SpanData;
+import datadog.trace.bootstrap.instrumentation.api.TagExtractor;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+
+/**
+ * Extracts the peer tags of a remote {@link InetSocketAddress}: those of its address (see {@link
+ * PeerAddress}), when it has one, and {@code peer.port}, when set.
+ */
+public final class PeerSocketAddress implements TagExtractor<InetSocketAddress> {
+  public static final PeerSocketAddress INSTANCE = new PeerSocketAddress();
+
+  private PeerSocketAddress() {}
+
+  @Override
+  public void extract(InetSocketAddress socketAddress, SpanData span) {
+    // an unresolved socket address is exactly one with no address
+    InetAddress address = socketAddress.getAddress();
+    if (address != null) {
+      PeerAddress.RESOLVED.extract(address, span);
+    }
+    int port = socketAddress.getPort();
+    if (port > BaseDecorator.UNSET_PORT) {
+      span.setTag(KnownTags.PEER_PORT_ID, port);
+    }
+  }
+}
@@ -0,0 +1,96 @@
+package datadog.trace.bootstrap.instrumentation.decorator;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import datadog.trace.api.KnownTags;
+import datadog.trace.bootstrap.instrumentation.api.SpanData;
+import datadog.trace.test.junit.utils.config.WithConfig;
+import datadog.trace.test.junit.utils.config.WithConfigExtension;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+// registered for the class so the one method-level @WithConfig is undone after that method
+@ExtendWith(WithConfigExtension.class)
+class PeerAddressTest {
+  private final SpanData span = mock(SpanData.class);
+
+  // getByAddress pairs a hostname with an IP without a DNS lookup
+  private static InetAddress ipv4() throws Exception {
+    return InetAddress.getByAddress("db.example", new byte[] {10, 0, 0, 1});
+  }
+
+  private static InetAddress ipv6() throws Exception {
+    byte[] ip = new byte[16];
+    ip[15] = 1;
+    return InetAddress.getByAddress("db6.example", ip);
+  }
+
+  @Test
+  void resolvedIpv4SetsHostnameAndIpv4() throws Exception {
+    PeerAddress.RESOLVED.extract(ipv4(), span);
+
+    verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db.example");
+    verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1");
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  void resolvedIpv6SetsHostnameAndIpv6() throws Exception {
+    InetAddress address = ipv6();
+    PeerAddress.RESOLVED.extract(address, span);
+
+    verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db6.example");
+    verify(span).setTag(KnownTags.PEER_IPV6_ID, (CharSequence) address.getHostAddress());
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  void unresolvedSkipsHostname() throws Exception {
+    PeerAddress.UNRESOLVED.extract(ipv4(), span);
+
+    verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1");
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  @WithConfig(key = "trace.peer.hostname.enabled", value = "false")
+  void hostnameDisabledSkipsHostname() throws Exception {
+    PeerAddress.RESOLVED.extract(ipv4(), span);
+
+    verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1");
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  void socketAddressAddsPort() throws Exception {
+    PeerSocketAddress.INSTANCE.extract(new InetSocketAddress(ipv4(), 5432), span);
+
+    verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db.example");
+    verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1");
+    verify(span).setTag(KnownTags.PEER_PORT_ID, 5432);
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  void unresolvedSocketAddressSetsOnlyPort() {
+    PeerSocketAddress.INSTANCE.extract(
+        InetSocketAddress.createUnresolved("bad.address.local", 999), span);
+
+    verify(span).setTag(KnownTags.PEER_PORT_ID, 999);
+    verifyNoMoreInteractions(span);
+  }
+
+  @Test
+  void unsetPortIsSkipped() throws Exception {
+    PeerSocketAddress.INSTANCE.extract(new InetSocketAddress(ipv4(), 0), span);
+
+    verify(span, never()).setTag(eq(KnownTags.PEER_PORT_ID), anyInt());
+  }
+}
@@ -66,6 +66,7 @@ public String[] helperClassNames() {
       packageName + ".TextMapInjectAdapterInterface",
       packageName + ".KafkaConsumerInfo",
       packageName + ".KafkaConsumerInstrumentationHelper",
+      packageName + ".KafkaConsumerTags",
       packageName + ".KafkaDecorator",
       packageName + ".TextMapExtractAdapter",
       packageName + ".TracingIterableDelegator",

@@ -0,0 +1,56 @@
+package datadog.trace.instrumentation.kafka_clients;
+
+import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP;
+import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS;
+import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID;
+import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME;
+
+import datadog.trace.bootstrap.instrumentation.api.FrozenTags;
+import datadog.trace.bootstrap.instrumentation.api.SpanData;
+import datadog.trace.bootstrap.instrumentation.api.TagExtractor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+/**
+ * The consumer tags every record of one topic in one poll shares -- destination, group, cluster id
+ * and bootstrap servers -- extracted from the topic and frozen once, then applied to each record's
+ * span. Records come out of a poll partition by partition, so the topic rarely changes between
+ * records and one cached topic suffices. Used by one iterator, on one thread.
+ */
+public final class KafkaConsumerTags implements TagExtractor<String> {
+  private final String group;
+  private final String clusterId;
+  private final String bootstrapServers;
+
+  private String frozenTopic;
+  private FrozenTags frozen;
+
+  public KafkaConsumerTags(String group, String clusterId, String bootstrapServers) {
+    this.group = group;
+    this.clusterId = clusterId;
+    this.bootstrapServers = bootstrapServers;
+  }
+
+  /** The frozen tags for {@code record}'s topic, frozen when the topic changes. */
+  public FrozenTags of(final ConsumerRecord<?, ?> record) {
+    final String topic = KafkaDecorator.topicOf(record);
+    if (topic != frozenTopic && !topic.equals(frozenTopic)) {
+      frozen = FrozenTags.of(topic, this);
+      frozenTopic = topic;
+    }
+    return frozen;
+  }
+
+  @Override
+  public void extract(final String topic, final SpanData span) {
+    span.setTag(MESSAGING_DESTINATION_NAME, topic);
+    if (group != null) {
+      span.setTag(CONSUMER_GROUP, group);
+    }
+    if (clusterId != null) {
+      span.setTag(KAFKA_CLUSTER_ID, clusterId);
+    }
+    if (bootstrapServers != null) {
+      span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers);
+    }
+  }
+}
@@ -1,6 +1,5 @@
 package datadog.trace.instrumentation.kafka_clients;
 
-import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP;
 import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS;
 import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID;
 import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME;
@@ -16,6 +15,7 @@
 import datadog.trace.api.cache.DDCaches;
 import datadog.trace.api.naming.SpanNaming;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.bootstrap.instrumentation.api.FrozenTags;
 import datadog.trace.bootstrap.instrumentation.api.InternalSpanTypes;
 import datadog.trace.bootstrap.instrumentation.api.Tags;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
@@ -114,27 +114,23 @@ protected String spanKind() {
     return spanKind;
   }
 
+  /** {@code record}'s topic, or {@code "kafka"} when it has none. */
+  static String topicOf(final ConsumerRecord<?, ?> record) {
+    return record.topic() == null ? "kafka" : record.topic();
+  }
+
+  /**
+   * Decorates a consumed record's span. {@code consumerTags} are the destination, group, cluster id
+   * and bootstrap servers, frozen once per topic per poll ({@link KafkaConsumerTags}).
+   */
   public void onConsume(
-      final AgentSpan span,
-      final ConsumerRecord record,
-      String consumerGroup,
-      String clusterId,
-      String bootstrapServers) {
+      final AgentSpan span, final ConsumerRecord record, final FrozenTags consumerTags) {
     if (record != null) {
-      final String topic = record.topic() == null ? "kafka" : record.topic();
+      final String topic = topicOf(record);
       span.setResourceName(CONSUMER_RESOURCE_NAME_CACHE.computeIfAbsent(topic, CONSUMER_PREFIX));
       span.setTag(PARTITION, record.partition());
       span.setTag(OFFSET, record.offset());
-      span.setTag(MESSAGING_DESTINATION_NAME, topic);
-      if (consumerGroup != null) {
-        span.setTag(CONSUMER_GROUP, consumerGroup);
-      }
-      if (clusterId != null) {
-        span.setTag(KAFKA_CLUSTER_ID, clusterId);
-      }
-      if (bootstrapServers != null) {
-        span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers);
-      }
+      span.setTagsFrom(consumerTags);
       // TODO - do we really need both? This mechanism already adds a lot of... baggage.
       // check to not record a duration if the message was sent from an old Kafka client
       if (record.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) {

@@ -46,7 +46,9 @@ public class TracingIterator implements Iterator<ConsumerRecord<?, ?>> {
   private final KafkaDecorator decorator;
   private final String group;
   private final String clusterId;
-  private final String bootstrapServers;
+
+  /** This poll's consumer tags, frozen per topic. */
+  private final KafkaConsumerTags consumerTags;
 
   public TracingIterator(
       final Iterator<ConsumerRecord<?, ?>> delegateIterator,
@@ -60,7 +62,7 @@ public TracingIterator(
     this.decorator = decorator;
     this.group = group;
     this.clusterId = clusterId;
-    this.bootstrapServers = bootstrapServers;
+    this.consumerTags = new KafkaConsumerTags(group, clusterId, bootstrapServers);
   }
 
   @Override
@@ -146,7 +148,7 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
           span.setTag(InstrumentationTags.TOMBSTONE, true);
         }
         decorator.afterStart(span);
-        decorator.onConsume(span, val, group, clusterId, bootstrapServers);
+        decorator.onConsume(span, val, consumerTags.of(val));
         if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) {
           activateNext(span);
         } else {