Capture Kafka consumer group membership on join - #11989
Capture Kafka consumer group membership on join#11989gh-worker-dd-mergequeue-cf854d[bot] merged 6 commits into
Conversation
Instrument ConsumerCoordinator.onJoinComplete (kafka-clients 0.11 and 3.8) to report the broker-assigned member id, generation id and negotiated member protocol each time a consumer (re)joins a group. Reported through Data Streams Monitoring alongside the consumer group and cluster id via a new reportKafkaConsumerGroupMember path (member_host is not available client-side and is intentionally omitted). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
piochelepiotr
added
tag: ai generated
This comment has been minimized.
This comment has been minimized.
🟢 Java Benchmark SLOs — All performance SLOs passed
PR vs. master results
Commit: Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion. |
Kafka / producer-benchmarkParameters
See matching parameters
SummaryFound 0 performance improvements and 0 performance regressions! Performance is the same for 3 metrics, 0 unstable metrics. See unchanged results
|
Kafka / consumer-benchmarkParameters
See matching parameters
SummaryFound 0 performance improvements and 0 performance regressions! Performance is the same for 3 metrics, 0 unstable metrics. See unchanged results
|
![datadog-datadog-prod-us1[bot]](https://avatars.githubusercontent.com/u/365230?s=60&v=4){kind=small}
There was a problem hiding this comment.
The PR instruments Kafka consumer group membership tracking but introduces a breaking msgpack format change: all kafka configs now serialize with 7 fields (adding MemberId, GenerationId, MemberProtocol) instead of 4. Downstream consumers that expect exactly 4 fields will fail to parse the new format, potentially breaking the entire DSM kafka pipeline. The PR claims cross-repo coordination, but without deployment verification, this is a critical downstream compatibility risk.
📊 Validated against 6 scenarios · Open Bits AI session
🤖 Datadog Autotest · Commit d299a2c · What is Autotest? · Any feedback? Reach out in #autotest
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d299a2cbf1
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
A membership report can't be attributed downstream without a cluster id, so skip it (the kafka_consumer integration reports the member with the cluster id and the tracer re-reports on the next rejoin) rather than emit an empty-cluster row. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
6e64884 to
9374995
Compare
…rship reporting Only mark a membership as reported once reportConsumerGroupMember actually attempts the report (i.e. cluster id is known), instead of marking it unconditionally before the cluster-id check. Also aligns generationId serialization with writeInt/unpackInt used elsewhere for int fields, and removes redundant/duplicated comments introduced by the feature. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
| } | ||
|
|
||
| @Nullable | ||
| public String getLastReportedMemberId() { |
There was a problem hiding this comment.
Minor: Similar to how the setLastReportedMembership() method sets both the memberId and the generationId in a single call, there could be a method to check if the membership has changed instead of using two separate getters.
…2103) Remove check for JAVA_TOOLS_OPTIONS in verifyAgentJarIntegrations Co-authored-by: sarah.chen <sarah.chen@datadoghq.com>
There was a problem hiding this comment.
When DSM is disabled during a consumer's initial join, the new helper returns success without sending the membership and the advice permanently records that tuple as reported. If DSM is enabled later, the member ID and generation remain absent until another rebalance changes the tuple, delaying the intended pod-to-partition correlation.
📊 Validated against 5 scenarios · Open Bits AI session
🤖 Datadog Autotest · Commit ef81ea3 · What is Autotest? · @DataDog review to ask questions · Any feedback? Reach out in #autotest
@datadog-datadog-prod-us1 I don't see this as a bug. DSM tracks member joins. If member joins happen outside of the time DSM is enabled, then we don't track them. |
|
/merge |
|
View all feedbacks in Devflow UI.
The expected merge time in
|
4 participants
What
Instruments
ConsumerCoordinator.onJoinComplete(kafka-clients 0.11 + 3.8) so that every time a consumer (re)joins a group the tracer reports its broker-assigned member id, generation id, and negotiated member protocol through Data Streams Monitoring, alongside the consumer group and Kafka cluster id.AgentDataStreamsMonitoring.reportKafkaConsumerGroupMember(...); carried on the DSM payload as first-class fields (MemberId,GenerationId,MemberProtocol) on the existing kafka config report — no new payload section.(memberId, generationId)avoids re-reporting unchanged membership.Motivation
The agent integration collects member ID --> [(topic, partition), ...] collection
However, users don't know what a member ID is. This PR adds the correlation member ID --> Infra tags (pod, host, etc).
That way, a user will be able to correlate a specific partition with a specific pod. (high lag on a partition & high CPU on the corresponding pod ==> probably a hot partition.