Skip to content

Capture kafka producer and consumer configs #10697

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gh-worker-dd-mergequeue-cf854d merged 12 commits into from
Apr 7, 2026
Merged

Capture kafka producer and consumer configs #10697

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
42d99ec
wip
piochelepiotr Dec 8, 2025
7ee74ff
wip
piochelepiotr Feb 17, 2026
b5faa7e
remove test application
piochelepiotr Feb 19, 2026
12584d7
update integration
piochelepiotr Feb 26, 2026
12ea586
Fix producer instrumentation FQN and use allowlist for config capture
piochelepiotr Mar 12, 2026
cb98198
Apply spotless formatting
piochelepiotr Mar 12, 2026
2499286
Fix Groovy evaluation order bug in DataStreamsWritingTest
piochelepiotr Mar 12, 2026
c0cff1d
Fix CodeNarc PackageName violation in KafkaConfigHelperTest
piochelepiotr Mar 13, 2026
fb55014
Fix review issues: type values, debug logging, dedup map, and null ch…
piochelepiotr Mar 17, 2026
7e9aff8
Fix context store type mismatch in producer context propagation advices
piochelepiotr Mar 18, 2026
a92256a
Use putIfAbsent for MetadataState context store to avoid double lookup
piochelepiotr Apr 1, 2026
bdbdf30
Use factory overload of putIfAbsent to avoid unnecessary MetadataStat…
piochelepiotr Apr 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions .../java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import datadog.trace.api.datastreams.DataStreamsTags;
import datadog.trace.bootstrap.InstrumentationContext;
import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.HashMap;
import java.util.Map;
import net.bytebuddy.asm.Advice;
Expand Down Expand Up @@ -42,7 +43,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
@Override
public Map<String, String> contextStore() {
Map<String, String> contextStores = new HashMap<>();
contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
contextStores.put(
Copy link
Copy Markdown
Contributor Author

@piochelepiotr piochelepiotr Mar 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I want to store both cluster ID & pending Kafka config under the same key. The way to do that is to create a container class.

Copy link
Copy Markdown
Contributor Author

@piochelepiotr piochelepiotr Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done - this is already handled by the MetadataState container class that holds both clusterId and pendingConfig.

"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
contextStores.put(
"org.apache.kafka.clients.consumer.internals.ConsumerCoordinator",
KafkaConsumerInfo.class.getName());
Expand All @@ -56,7 +59,11 @@ public String instrumentedType() {

@Override
public String[] helperClassNames() {
return new String[] {packageName + ".KafkaConsumerInfo"};
return new String[] {
packageName + ".KafkaConsumerInfo",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

@Override
Expand Down Expand Up @@ -90,7 +97,9 @@ public static void trackCommitOffset(
Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
String clusterId = null;
if (consumerMetadata != null) {
clusterId = InstrumentationContext.get(Metadata.class, String.class).get(consumerMetadata);
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
clusterId = metadataState != null ? metadataState.clusterId : null;
}

for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
Expand Down
34 changes: 31 additions & 3 deletions ...in/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfoInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import datadog.trace.bootstrap.instrumentation.api.AgentScope;
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
import datadog.trace.instrumentation.kafka_common.ClusterIdHolder;
import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -60,7 +62,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
@Override
public Map<String, String> contextStore() {
Map<String, String> contextStores = new HashMap<>();
contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
contextStores.put(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
contextStores.put(
"org.apache.kafka.clients.consumer.ConsumerRecords", KafkaConsumerInfo.class.getName());
contextStores.put(
Expand All @@ -82,6 +86,9 @@ public String[] helperClassNames() {
packageName + ".KafkaDecorator",
packageName + ".KafkaConsumerInfo",
"datadog.trace.instrumentation.kafka_common.ClusterIdHolder",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

Expand Down Expand Up @@ -148,6 +155,16 @@ public static void captureGroup(
.put(coordinator, kafkaConsumerInfo);
}
}

if (Config.get().isDataStreamsEnabled()) {
MetadataState state =
InstrumentationContext.get(Metadata.class, MetadataState.class)
.putIfAbsent(metadata, MetadataState::new);
KafkaConfigHelper.storePendingConsumerConfig(
state,
normalizedConsumerGroup,
KafkaConfigHelper.extractConsumerConfig(consumerConfig));
}
}

public static void muzzleCheck(ConsumerRecord record) {
Expand Down Expand Up @@ -191,6 +208,16 @@ public static void captureGroup(
.put(coordinator, kafkaConsumerInfo);
}
}

if (Config.get().isDataStreamsEnabled()) {
MetadataState state =
InstrumentationContext.get(Metadata.class, MetadataState.class)
.putIfAbsent(metadata, MetadataState::new);
KafkaConfigHelper.storePendingConsumerConfig(
state,
normalizedConsumerGroup,
KafkaConfigHelper.extractConsumerConfigFromMap(consumerConfig));
}
}

public static void muzzleCheck(ConsumerRecord record) {
Expand All @@ -214,8 +241,9 @@ public static AgentScope onEnter(@Advice.This KafkaConsumer consumer) {
if (kafkaConsumerInfo != null && Config.get().isDataStreamsEnabled()) {
Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
if (consumerMetadata != null) {
String clusterId =
InstrumentationContext.get(Metadata.class, String.class).get(consumerMetadata);
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
String clusterId = metadataState != null ? metadataState.clusterId : null;
if (clusterId != null) {
ClusterIdHolder.set(clusterId);
}
Expand Down
13 changes: 9 additions & 4 deletions ...c/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.agent.tooling.InstrumenterModule;
import datadog.trace.bootstrap.InstrumentationContext;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
Expand Down Expand Up @@ -46,7 +47,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
@Override
public Map<String, String> contextStore() {
Map<String, String> contextStores = new HashMap<>(2);
contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
contextStores.put(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
contextStores.put(
"org.apache.kafka.clients.consumer.ConsumerRecords", KafkaConsumerInfo.class.getName());
return contextStores;
Expand All @@ -73,6 +76,8 @@ public String[] helperClassNames() {
packageName + ".TextMapInjectAdapter",
"datadog.trace.instrumentation.kafka_common.Utils",
"datadog.trace.instrumentation.kafka_common.StreamingContext",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

Expand Down Expand Up @@ -113,7 +118,7 @@ public static void wrap(
String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
String clusterId =
KafkaConsumerInstrumentationHelper.extractClusterId(
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
String bootstrapServers =
KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
iterable =
Expand All @@ -135,7 +140,7 @@ public static void wrap(
String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
String clusterId =
KafkaConsumerInstrumentationHelper.extractClusterId(
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
String bootstrapServers =
KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
iterable =
Expand All @@ -157,7 +162,7 @@ public static void wrap(
String group = KafkaConsumerInstrumentationHelper.extractGroup(kafkaConsumerInfo);
String clusterId =
KafkaConsumerInstrumentationHelper.extractClusterId(
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, String.class));
kafkaConsumerInfo, InstrumentationContext.get(Metadata.class, MetadataState.class));
String bootstrapServers =
KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo);
iterator =
Expand Down
7 changes: 5 additions & 2 deletions .../java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentationHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import datadog.trace.api.Config;
import datadog.trace.bootstrap.ContextStore;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import org.apache.kafka.clients.Metadata;

public class KafkaConsumerInstrumentationHelper {
Expand All @@ -13,11 +14,13 @@ public static String extractGroup(KafkaConsumerInfo kafkaConsumerInfo) {
}

public static String extractClusterId(
KafkaConsumerInfo kafkaConsumerInfo, ContextStore<Metadata, String> metadataContextStore) {
KafkaConsumerInfo kafkaConsumerInfo,
ContextStore<Metadata, MetadataState> metadataContextStore) {
if (Config.get().isDataStreamsEnabled() && kafkaConsumerInfo != null) {
Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
if (consumerMetadata != null) {
return metadataContextStore.get(consumerMetadata);
MetadataState state = metadataContextStore.get(consumerMetadata);
return state != null ? state.clusterId : null;
}
}
return null;
Expand Down
40 changes: 37 additions & 3 deletions ...c/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import static datadog.trace.instrumentation.kafka_common.StreamingContext.STREAMING_CONTEXT;
import static datadog.trace.instrumentation.kafka_common.Utils.DSM_TRANSACTION_SOURCE_READER;
import static java.util.Collections.singletonMap;
import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
import static net.bytebuddy.matcher.ElementMatchers.isPrivate;
import static net.bytebuddy.matcher.ElementMatchers.isPublic;
Expand All @@ -42,6 +43,8 @@
import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags;
import datadog.trace.instrumentation.kafka_common.ClusterIdHolder;
import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.Map;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.matcher.ElementMatcher;
Expand Down Expand Up @@ -88,17 +91,29 @@ public String[] helperClassNames() {
"datadog.trace.instrumentation.kafka_common.StreamingContext",
"datadog.trace.instrumentation.kafka_common.ClusterIdHolder",
"datadog.trace.instrumentation.kafka_common.Utils",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
packageName + ".AvroSchemaExtractor",
};
}

@Override
public Map<String, String> contextStore() {
return singletonMap("org.apache.kafka.clients.Metadata", "java.lang.String");
return singletonMap(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
}

@Override
public void methodAdvice(MethodTransformer transformer) {
transformer.applyAdvice(
isConstructor()
.and(takesArgument(0, named("org.apache.kafka.clients.producer.ProducerConfig")))
.and(takesArgument(1, named("org.apache.kafka.common.serialization.Serializer")))
.and(takesArgument(2, named("org.apache.kafka.common.serialization.Serializer"))),
KafkaProducerInstrumentation.class.getName() + "$ProducerConstructorAdvice");

transformer.applyAdvices(
isMethod()
.and(isPublic())
Expand Down Expand Up @@ -126,7 +141,9 @@ public static AgentScope onEnter(
@Advice.FieldValue("metadata") Metadata metadata,
@Advice.Argument(value = 0, readOnly = false) ProducerRecord record,
@Advice.Argument(value = 1, readOnly = false) Callback callback) {
String clusterId = InstrumentationContext.get(Metadata.class, String.class).get(metadata);
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(metadata);
String clusterId = metadataState != null ? metadataState.clusterId : null;

// Set cluster ID for Schema Registry instrumentation
if (clusterId != null) {
Expand Down Expand Up @@ -183,7 +200,9 @@ public static void onEnter(
@Advice.Argument(value = 0, readOnly = false) ProducerRecord record) {
AgentSpan span = activeSpan();
if (span == null) return;
String clusterId = InstrumentationContext.get(Metadata.class, String.class).get(metadata);
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(metadata);
String clusterId = metadataState != null ? metadataState.clusterId : null;
TextMapInjectAdapterInterface setter = NoopTextMapInjectAdapter.NOOP_SETTER;
// Do not inject headers for batch versions below 2
// This is how similar check is being done in Kafka client itself:
Expand Down Expand Up @@ -243,6 +262,21 @@ record =
}
}

public static class ProducerConstructorAdvice {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void captureConfiguration(
@Advice.FieldValue("metadata") Metadata metadata,
@Advice.Argument(0) ProducerConfig producerConfig) {
if (Config.get().isDataStreamsEnabled()) {
MetadataState state =
InstrumentationContext.get(Metadata.class, MetadataState.class)
.putIfAbsent(metadata, MetadataState::new);
KafkaConfigHelper.storePendingProducerConfig(
state, KafkaConfigHelper.extractProducerConfig(producerConfig));
}
}
}

public static class PayloadSizeAdvice {

/**
Expand Down
29 changes: 23 additions & 6 deletions ...11/src/main/java/datadog/trace/instrumentation/kafka_clients/MetadataInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.agent.tooling.InstrumenterModule;
import datadog.trace.bootstrap.InstrumentationContext;
import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.Map;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.type.TypeDescription;
Expand Down Expand Up @@ -51,12 +53,19 @@ public ElementMatcher<TypeDescription> hierarchyMatcher() {

@Override
public String[] helperClassNames() {
return new String[] {packageName + ".KafkaDecorator"};
return new String[] {
packageName + ".KafkaDecorator",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

@Override
public Map<String, String> contextStore() {
return singletonMap("org.apache.kafka.clients.Metadata", "java.lang.String");
return singletonMap(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
}

@Override
Expand All @@ -78,8 +87,12 @@ public static class MetadataUpdateBefore22Advice {
public static void onEnter(
@Advice.This final Metadata metadata, @Advice.Argument(0) final Cluster newCluster) {
if (newCluster != null && !newCluster.isBootstrapConfigured()) {
InstrumentationContext.get(Metadata.class, String.class)
.put(metadata, newCluster.clusterResource().clusterId());
String clusterId = newCluster.clusterResource().clusterId();
MetadataState state =
InstrumentationContext.get(Metadata.class, MetadataState.class)
.putIfAbsent(metadata, MetadataState::new);
state.clusterId = clusterId;
KafkaConfigHelper.reportPendingConfig(state, clusterId);
}
}

Expand All @@ -95,8 +108,12 @@ public static class MetadataUpdate22AndAfterAdvice {
public static void onEnter(
@Advice.This final Metadata metadata, @Advice.Argument(1) final MetadataResponse response) {
if (response != null) {
InstrumentationContext.get(Metadata.class, String.class)
.put(metadata, response.clusterId());
String clusterId = response.clusterId();
MetadataState state =
InstrumentationContext.get(Metadata.class, MetadataState.class)
.putIfAbsent(metadata, MetadataState::new);
state.clusterId = clusterId;
KafkaConfigHelper.reportPendingConfig(state, clusterId);
}
}

Expand Down
10 changes: 8 additions & 2 deletions ...ava/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
@Override
public Map<String, String> contextStore() {
Map<String, String> contextStores = new HashMap<>(2);
contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
contextStores.put(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
contextStores.put(
"org.apache.kafka.clients.consumer.internals.ConsumerCoordinator",
KafkaConsumerInfo.class.getName());
Expand All @@ -41,7 +43,11 @@ public String instrumentedType() {

@Override
public String[] helperClassNames() {
return new String[] {packageName + ".KafkaConsumerInfo"};
return new String[] {
packageName + ".KafkaConsumerInfo",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

@Override
Expand Down
7 changes: 6 additions & 1 deletion .../java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfoInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,9 @@ public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
@Override
public Map<String, String> contextStore() {
Map<String, String> contextStores = new HashMap<>(4);
contextStores.put("org.apache.kafka.clients.Metadata", "java.lang.String");
contextStores.put(
"org.apache.kafka.clients.Metadata",
"datadog.trace.instrumentation.kafka_common.MetadataState");
contextStores.put(
"org.apache.kafka.clients.consumer.ConsumerRecords", KafkaConsumerInfo.class.getName());
// new- here we are storing the callbackinvoker and consumerdelegate in the context store
Expand Down Expand Up @@ -77,6 +79,9 @@ public String[] helperClassNames() {
packageName + ".KafkaConsumerInfo",
packageName + ".KafkaConsumerInstrumentationHelper",
"datadog.trace.instrumentation.kafka_common.ClusterIdHolder",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
}

Expand Down
Loading
Loading