Capture Kafka consumer group membership on join#11989
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 1 performance regressions! Performance is the same for 2 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
| packer.writeUTF8(CONFIG_GENERATION_ID); | ||
| packer.writeLong(config.getGenerationId()); | ||
|
|
||
| packer.writeUTF8(CONFIG_MEMBER_PROTOCOL); |
There was a problem hiding this comment.
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
![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".
| transformer.applyAdvice( | ||
| isMethod().and(named("onJoinComplete")).and(takesArguments(4)), | ||
| packageName + ".JoinGroupAdvice"); |
There was a problem hiding this comment.
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 👍 / 👎.
| && generationId == kafkaConsumerInfo.getLastReportedGenerationId()) { | ||
| return; | ||
| } | ||
| kafkaConsumerInfo.setLastReportedMembership(memberId, generationId); |
There was a problem hiding this comment.
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 👍 / 👎.
1 participant
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.member_hostis intentionally omitted: it is assigned broker-side and never sent to the joining client.Part of a cross-repo change
This is the producer side. Downstream: dd-go data-pipeline-edge →
dsm-kafka-configsstream → dsm-kafka-configs-writer → orgstoreconsumer_group_memberstable → dsm-api.Tests
DataStreamsWritingTest+DefaultDataStreamsMonitoringTest(new member/generation/protocol wire-format assertions) — green.🤖 Generated with Claude Code