Skip to content

Capture Kafka consumer group membership on join #11989

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

Open
piochelepiotr wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Capture Kafka consumer group membership on join #11989

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
47 changes: 47 additions & 0 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.KafkaConfigHelper;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import java.util.HashMap;
import java.util.Map;
Expand Down Expand Up @@ -61,6 +62,7 @@ public String instrumentedType() {
public String[] helperClassNames() {
return new String[] {
packageName + ".KafkaConsumerInfo",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
Expand All @@ -71,6 +73,9 @@ public void methodAdvice(MethodTransformer transformer) {
transformer.applyAdvice(
isMethod().and(named("sendOffsetCommitRequest")).and(takesArguments(1)),
ConsumerCoordinatorInstrumentation.class.getName() + "$CommitOffsetAdvice");
transformer.applyAdvice(
isMethod().and(named("onJoinComplete")).and(takesArguments(4)),
ConsumerCoordinatorInstrumentation.class.getName() + "$JoinGroupAdvice");
}

public static class CommitOffsetAdvice {
Expand Down Expand Up @@ -127,4 +132,46 @@ public static void muzzleCheck(ConsumerRecord record) {
record.headers();
}
}

public static class JoinGroupAdvice {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void trackJoinGroup(
@Advice.This ConsumerCoordinator coordinator,
@Advice.Argument(0) final int generationId,
@Advice.Argument(1) final String memberId,
@Advice.Argument(2) final String memberProtocol) {
if (memberId == null || memberId.isEmpty()) {
return;
}
KafkaConsumerInfo kafkaConsumerInfo =
InstrumentationContext.get(ConsumerCoordinator.class, KafkaConsumerInfo.class)
.get(coordinator);
if (kafkaConsumerInfo == null) {
return;
}
// Only report when the membership changes (new member id or new generation) to avoid
// re-reporting an unchanged membership.
if (memberId.equals(kafkaConsumerInfo.getLastReportedMemberId())
&& generationId == kafkaConsumerInfo.getLastReportedGenerationId()) {
return;
}
kafkaConsumerInfo.setLastReportedMembership(memberId, generationId);

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Defer reporting until the cluster ID is known

If a consumer completes its first join before MetadataState.clusterId has been populated, this records the (memberId, generationId) as already reported and then sends a membership report with an empty cluster id. When the later metadata update fills in the cluster id, there is no rejoin and the guard above suppresses re-emitting the same membership, unlike the existing config path which stores pending reports until cluster id is available. This leaves downstream member rows permanently missing the Kafka cluster id for that startup ordering; consider keeping the membership pending or only updating lastReportedMembership after a report with a known cluster id.

Useful? React with 👍 / 👎.


String consumerGroup = kafkaConsumerInfo.getConsumerGroup();
Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata();
String clusterId = null;
if (consumerMetadata != null) {
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
clusterId = metadataState != null ? metadataState.clusterId : null;
}
KafkaConfigHelper.reportConsumerGroupMember(
clusterId, consumerGroup, memberId, generationId, memberProtocol);
}

public static void muzzleCheck(ConsumerRecord record) {
// Match CommitOffsetAdvice: only apply for kafka versions with headers
record.headers();
}
}
}
18 changes: 18 additions & 0 deletions ...nts-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@ public class KafkaConsumerInfo {
private final String consumerGroup;
private final Metadata clientMetadata;
private final String bootstrapServers;
// Last consumer group membership reported to DSM; used to avoid re-reporting when neither the
// member id nor the generation changed. Not part of identity (excluded from equals/hashCode).
private volatile String lastReportedMemberId;
private volatile int lastReportedGenerationId = Integer.MIN_VALUE;

public KafkaConsumerInfo(String consumerGroup, Metadata clientMetadata, String bootstrapServers) {
this.consumerGroup = consumerGroup;
Expand Down Expand Up @@ -36,6 +40,20 @@ public String getBootstrapServers() {
return bootstrapServers;
}

@Nullable
public String getLastReportedMemberId() {
return lastReportedMemberId;
}

public int getLastReportedGenerationId() {
return lastReportedGenerationId;
}

public void setLastReportedMembership(String memberId, int generationId) {
this.lastReportedMemberId = memberId;
this.lastReportedGenerationId = generationId;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
Expand Down
4 changes: 4 additions & 0 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 @@ -45,6 +45,7 @@ public String instrumentedType() {
public String[] helperClassNames() {
return new String[] {
packageName + ".KafkaConsumerInfo",
"datadog.trace.instrumentation.kafka_common.KafkaConfigHelper",
"datadog.trace.instrumentation.kafka_common.PendingConfig",
"datadog.trace.instrumentation.kafka_common.MetadataState",
};
Expand All @@ -55,5 +56,8 @@ public void methodAdvice(MethodTransformer transformer) {
transformer.applyAdvice(
isMethod().and(named("sendOffsetCommitRequest")).and(takesArguments(1)),
packageName + ".ConsumerCoordinatorAdvice");
transformer.applyAdvice(
isMethod().and(named("onJoinComplete")).and(takesArguments(4)),
packageName + ".JoinGroupAdvice");
Comment on lines +59 to +61

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Instrument async 3.8 consumers too

With Kafka 3.8+ consumers configured with group.protocol=consumer (the KIP-848 path), Kafka constructs the async ConsumerDelegate path rather than the legacy coordinator-backed consumer; this module already reflects that by storing KafkaConsumerInfo on ConsumerDelegate/OffsetCommitCallbackInvoker in ConstructorAdvice instead of on ConsumerCoordinator. Because the new member reporting is only attached to ConsumerCoordinator.onJoinComplete, those supported 3.8 consumers never execute this advice and no member/generation/protocol report is emitted for them. Please add equivalent reporting on the async membership path or explicitly restrict the feature to the legacy protocol.

Useful? React with 👍 / 👎.

}
}
51 changes: 51 additions & 0 deletions ...ts-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/JoinGroupAdvice.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
package datadog.trace.instrumentation.kafka_clients38;

import datadog.trace.bootstrap.InstrumentationContext;
import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper;
import datadog.trace.instrumentation.kafka_common.MetadataState;
import net.bytebuddy.asm.Advice;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;

public class JoinGroupAdvice {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void trackJoinGroup(
@Advice.This ConsumerCoordinator coordinator,
@Advice.Argument(0) final int generationId,
@Advice.Argument(1) final String memberId,
@Advice.Argument(2) final String memberProtocol) {
if (memberId == null || memberId.isEmpty()) {
return;
}
KafkaConsumerInfo kafkaConsumerInfo =
InstrumentationContext.get(ConsumerCoordinator.class, KafkaConsumerInfo.class)
.get(coordinator);
if (kafkaConsumerInfo == null) {
return;
}
// Only report when the membership changes (new member id or new generation) to avoid
// re-reporting an unchanged membership.
if (memberId.equals(kafkaConsumerInfo.getLastReportedMemberId().orElse(null))
&& generationId == kafkaConsumerInfo.getLastReportedGenerationId()) {
return;
}
kafkaConsumerInfo.setLastReportedMembership(memberId, generationId);

String consumerGroup = kafkaConsumerInfo.getConsumerGroup().orElse(null);
Metadata consumerMetadata = kafkaConsumerInfo.getmetadata().orElse(null);
String clusterId = null;
if (consumerMetadata != null) {
MetadataState metadataState =
InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata);
clusterId = metadataState != null ? metadataState.clusterId : null;
}
KafkaConfigHelper.reportConsumerGroupMember(
clusterId, consumerGroup, memberId, generationId, memberProtocol);
}

public static void muzzleCheck(ConsumerRecord record) {
// Match ConsumerCoordinatorAdvice: only apply for kafka versions with headers
record.headers();
}
}
17 changes: 17 additions & 0 deletions ...-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@ public class KafkaConsumerInfo {
private final String consumerGroup;
private final Metadata metadata;
private final String bootstrapServers;
// Last consumer group membership reported to DSM; used to avoid re-reporting when neither the
// member id nor the generation changed. Not part of identity (excluded from equals/hashCode).
private volatile String lastReportedMemberId;
private volatile int lastReportedGenerationId = Integer.MIN_VALUE;

public KafkaConsumerInfo(String consumerGroup, Metadata metadata, String bootstrapServers) {
this.consumerGroup = consumerGroup;
Expand All @@ -33,6 +37,19 @@ public Optional<String> getBootstrapServers() {
return Optional.ofNullable(bootstrapServers);
}

public Optional<String> getLastReportedMemberId() {
return Optional.ofNullable(lastReportedMemberId);
}

public int getLastReportedGenerationId() {
return lastReportedGenerationId;
}

public void setLastReportedMembership(String memberId, int generationId) {
this.lastReportedMemberId = memberId;
this.lastReportedGenerationId = generationId;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
Expand Down
25 changes: 25 additions & 0 deletions ...ka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,31 @@ public static void storePendingConsumerConfig(
log.debug("Stored pending consumer config (cluster ID not yet known)");
}

/**
* Reports consumer group membership when a consumer (re)joins a group. The broker-assigned member
* id is sent alongside the cluster id and consumer group.
*/
public static void reportConsumerGroupMember(
String clusterId,
String consumerGroup,
String memberId,
int generationId,
String memberProtocol) {
if (memberId == null || memberId.isEmpty()) {
return;
}
if (Config.get().isDataStreamsEnabled()) {
AgentTracer.get()
.getDataStreamsMonitoring()
.reportKafkaConsumerGroupMember(
clusterId != null ? clusterId : "",
consumerGroup != null ? consumerGroup : "",
memberId,
generationId,
memberProtocol != null ? memberProtocol : "");
}
}

/** Called from metadata update advice when the cluster ID becomes available. */
public static void reportPendingConfig(MetadataState state, String clusterId) {
PendingConfig pending = state.takePendingConfig();
Expand Down
20 changes: 20 additions & 0 deletions dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,26 @@ public void reportKafkaConfig(
getThreadServiceName()));
}

@Override
public void reportKafkaConsumerGroupMember(
String kafkaClusterId,
String consumerGroup,
String memberId,
int generationId,
String memberProtocol) {
inbox.offer(
new KafkaConfigReport(
"kafka_consumer",
kafkaClusterId,
consumerGroup,
memberId,
generationId,
memberProtocol,
Collections.<String, String>emptyMap(),
timeSource.getCurrentTimeNanos(),
getThreadServiceName()));
}

@Override
public void setCheckpoint(AgentSpan span, DataStreamsContext context) {
PathwayContext pathwayContext = span.spanContext().getPathwayContext();
Expand Down
15 changes: 14 additions & 1 deletion ...ce-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ public class MsgPackDatastreamsPayloadWriter implements DatastreamsPayloadWriter
private static final byte[] CONFIG_TYPE = "Type".getBytes(ISO_8859_1);
private static final byte[] CONFIG_KAFKA_CLUSTER_ID = "KafkaClusterId".getBytes(ISO_8859_1);
private static final byte[] CONFIG_CONSUMER_GROUP = "ConsumerGroup".getBytes(ISO_8859_1);
private static final byte[] CONFIG_MEMBER_ID = "MemberId".getBytes(ISO_8859_1);
private static final byte[] CONFIG_GENERATION_ID = "GenerationId".getBytes(ISO_8859_1);
private static final byte[] CONFIG_MEMBER_PROTOCOL = "MemberProtocol".getBytes(ISO_8859_1);
private static final byte[] CONFIG_ENTRIES = "Config".getBytes(ISO_8859_1);

private static final int INITIAL_CAPACITY = 512 * 1024;
Expand Down Expand Up @@ -290,7 +293,8 @@ private void writeKafkaConfigs(List<KafkaConfigReport> configs, Writable packer)
packer.writeUTF8(CONFIGS);
packer.startArray(configs.size());
for (KafkaConfigReport config : configs) {
packer.startMap(4); // Type, KafkaClusterId, ConsumerGroup, Config
// Type, KafkaClusterId, ConsumerGroup, MemberId, GenerationId, MemberProtocol, Config
packer.startMap(7);

packer.writeUTF8(CONFIG_TYPE);
packer.writeString(config.getType(), null);
Expand All @@ -301,6 +305,15 @@ private void writeKafkaConfigs(List<KafkaConfigReport> configs, Writable packer)
packer.writeUTF8(CONFIG_CONSUMER_GROUP);
packer.writeString(config.getConsumerGroup(), null);

packer.writeUTF8(CONFIG_MEMBER_ID);
packer.writeString(config.getMemberId(), null);

packer.writeUTF8(CONFIG_GENERATION_ID);
packer.writeLong(config.getGenerationId());

packer.writeUTF8(CONFIG_MEMBER_PROTOCOL);

@datadog-datadog-prod-us1 datadog-datadog-prod-us1 Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

P1 Breaking msgpack format change for all kafka configs

DSM kafka configuration pipeline may fail to parse reports, leading to data loss and broken consumer group monitoring. This breaks production observability for Kafka consumers using Data Streams Monitoring.

Assertion details
  • Input: Any system downstream of DSM that consumes Kafka config reports (dsm-kafka-configs stream, dsm-kafka-configs-writer, etc.) and expects exactly 4 fields in the msgpack map.
  • Expected: Kafka config reports serialize with 4 fields: Type, KafkaClusterId, ConsumerGroup, Config. Downstream code that calls unpackMapHeader() expecting 4 and then reads exactly 4 fields should succeed.
  • Actual: Kafka config reports now serialize with 7 fields: Type, KafkaClusterId, ConsumerGroup, MemberId, GenerationId, MemberProtocol, Config. Any downstream unpacker expecting 4 fields will read 7 and either: (1) fail with assertion error, (2) read wrong data beyond the 4 expected fields, or (3) silently ignore the 3 new fields if using lenient parsing.

Was this helpful? React 👍 or 👎
🤖 Datadog Autotest · What is Autotest? · Any feedback? Reach out in #autotest

packer.writeString(config.getMemberProtocol(), null);

packer.writeUTF8(CONFIG_ENTRIES);
Map<String, String> entries = config.getConfig();
packer.startMap(entries.size());
Expand Down
Loading
Loading