Add consumer group sharding to kafka_consumer by dsavov · Pull Request #25886 · DataDog/integrations-core

dsavov · 2026-10-31T07:39:58Z

What does this PR do?

Adds two kafka_consumer instance options, consumer_groups_shard_count and consumer_groups_shard_index, to split the consumer groups of one Kafka cluster across several instances.

Each group belongs to shard zlib.crc32(group_id) % consumer_groups_shard_count, so every instance agrees on the assignment across processes and restarts. _get_consumer_groups drops the groups outside the instance's shard before any offsets are fetched, with monitor_unlisted_consumer_groups, consumer_groups_regex and consumer_groups alike. Highwater offsets follow, as they are only fetched for the partitions the shard's groups consume.

consumer_groups_shard_index is required when the count is above 1: defaulting it to 0 would silently collect only part of the groups. Out-of-range values raise ConfigurationError, and so does sharding together with enable_kafka_console, whose cluster-wide metadata collection would otherwise run on every shard. Without the options nothing changes.

Motivation

With monitor_unlisted_consumer_groups, one check run collects every group of the cluster, and that run can only go to one cluster check runner. On a cluster with about 17k groups (about 1.08M consumer contexts), a single run took about 30s at 1.8 cores and 1.6Gi, and still stopped at the 2M context limit. On a cluster with about 42k groups, runners were OOM-killed at 16Gi.

Running this branch against the 17k-group cluster as 3 instances on 3 runners, each instance covered between 5,731 and 5,831 groups with no overlap, ran in 6 to 9s and peaked at about 0.7Gi.

Two Cluster Agent changes build on this: DataDog/datadog-agent#59382 generates the shard instances from a single config, and DataDog/datadog-agent#59381 keeps them on different runners.

QA: kafka_consumer is only owned by @DataDog/agent-integrations, so this should be qa/skip-qa. I can't set labels from a fork.

Checklist before requesting review

I self-reviewed the full diff, and this description matches the final change
Tests cover the change, or this description explains why none are needed
Automated review comments, including Codex review, were addressed or answered
I, the human author, understand this change, can explain it to reviewers, and believe it meets the repository standards

dd-octo-sts · 2026-10-31T07:41:40Z

✅ Dispatcher tests: passed

Dispatcher beta: informational only
Existing CI remains the merge signal.

6/6 jobs

✅ 6 passed · nothing failed

Batches · ✅ batch-01 6/6

_{Dispatcher finished on 627c14f — GitHub Run · Dispatcher Logs.}

datadog-datadog-us1-prod · 2026-10-31T07:42:02Z

❌ Errors

Your PR has failed checks. Please review the issues below and take necessary action before merging.

🚦 1 Pipeline job failed

Validate repository | Run Validations / Validate

View more details · View in GitHub Actions

ℹ️ Info

🎯 Code Coverage (details)
• Patch Coverage: 100.00%
• Overall Coverage: 92.95% (+3.19%)

Useful? React with 👍 / 👎

_{This comment will be updated automatically if new data arrives.

🔗 Commit SHA: 44ba95a | Docs | View more details | Give us feedback!}

dsavov · 2026-10-31T08:27:37Z

I have read the CLA Document and I hereby sign the CLA

dsavov · 2026-10-31T08:31:31Z

I have read the CLA Document and I hereby sign the CLA

chatgpt-codex-connector · 2026-10-31T08:56:54Z

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-31T09:37:30.080732Z	`44ba95a`	Manual request

ℹ️ About Codex in GitHub

Your team has set up Codex to 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" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

datadog-datadog-us1-prod

Combining sharding with Kafka Console counts each consumer-group membership change once per shard because metadata collection ignores shard ownership.

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit f506c7f}

datadog-datadog-us1-prod · 2026-10-31T09:00:51Z

+        in_shard = self.config.in_consumer_groups_shard
        if self.config._monitor_unlisted_consumer_groups:
-            return [grp for grp in self.client.list_consumer_groups() if grp]
+            return [grp for grp in self.client.list_consumer_groups() if grp and in_shard(grp)]


Apply shard ownership to Kafka Console group metadata

With enable_kafka_console enabled across N shard instances, this filter partitions offset collection, but ClusterMetadataCollector independently describes every consumer group on every instance. Each membership change therefore emits kafka.consumer_group.membership_changes once per shard instead of once overall, inflating counts and duplicating membership events. Apply shard ownership to consumer-group metadata collection as well, or reject this configuration combination.

React: 👍 useful · 👎 wrong · 😕 unclear · 👀 right, but not acting on it

What each reaction means

👍 Useful: correct, and you're fixing it or will.

👎 Wrong: the finding is incorrect, for example the bug can't happen or the code was misread.

😕 Unclear: you can't tell what it means or why it matters.

👀 Right, not acting: correct, but intended, out of scope for this PR, a known risk, or too minor.

_{🤖 Bits Code Review} _{· Open Bits AI session}

Agreed. Fixed in fbb9a231 by rejecting the combination: consumer_groups_shard_count greater than 1 with enable_kafka_console (or enable_cluster_monitoring) now raises a ConfigurationError, with a test for both option names.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f506c7f438

ℹ️ 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".

chatgpt-codex-connector · 2026-10-31T09:02:55Z

+        in_shard = self.config.in_consumer_groups_shard
        if self.config._monitor_unlisted_consumer_groups:
-            return [grp for grp in self.client.list_consumer_groups() if grp]
+            return [grp for grp in self.client.list_consumer_groups() if grp and in_shard(grp)]


Shard Kafka Console group metadata too

When enable_kafka_console is enabled (which sets _monitor_unlisted_consumer_groups), this filters only the offset-discovery path. The same run invokes ClusterMetadataCollector.collect_all_metadata, and _collect_consumer_group_metadata in cluster_metadata.py:782-858 still lists, describes, and emits metrics/events for every group on every shard. Consequently an N-shard deployment retains the full metadata workload on each runner and can emit consumer_group.membership_changes once per shard with identical tags for a single transition, overcounting that metric; apply the shard predicate to this collector while keeping any truly cluster-wide summary on one shard.

Useful? React with 👍 / 👎.

Good catch, thanks. Fixed in fbb9a231 by rejecting the combination: an instance with consumer_groups_shard_count greater than 1 and enable_kafka_console (or enable_cluster_monitoring) now raises a ConfigurationError, and the option docs say so. Sharding the Kafka Console metadata, with the cluster-wide part on a single shard, would be a larger change to ClusterMetadataCollector; I would rather keep it out of this PR.

dsavov · 2026-10-31T09:14:53Z

Sharding is currently rejected together with enable_kafka_console, since console collects cluster-wide metadata (brokers, topics, schemas, all groups, heartbeat) on every run. If you'd like console support, I can follow up by sharding the per-group metadata and collecting the cluster-wide parts on shard 0 only, but I'd need guidance on whether the console backend accepts group metadata and membership events from several instances of one cluster, and which instance should own the heartbeat. Happy to do it here or in a separate PR, whichever you prefer.

dsavov · 2026-10-31T09:16:33Z

@codex review

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: fbb9a23158

ℹ️ 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".

chatgpt-codex-connector · 2026-10-31T09:20:29Z

+
+
+def sharded_instance(kafka_instance, shard_index, shard_count=3, **options):
+    kafka_instance.pop("consumer_groups", None)


Add type hints to the new test helper

The newly added sharded_instance helper leaves its arguments and return value unannotated even though it is a regular helper function, so add annotations for the instance mapping, shard values, options, and returned mapping to comply with the root repository requirement that new functions and methods carry type hints.

AGENTS.md reference: AGENTS.md:L33-L35

Useful? React with 👍 / 👎.

Fixed in 44ba95a: sharded_instance now annotates its arguments and return type.

Add consumer_groups_shard_count and consumer_groups_shard_index so one cluster's consumer groups can be split across several instances, and therefore several cluster check runners. Groups are assigned by a stable crc32 hash of the group ID and dropped before their offsets are fetched.

dsavov · 2026-10-31T09:33:02Z

@codex review

chatgpt-codex-connector · 2026-10-31T09:37:28Z

Codex Review: Didn't find any major issues. Hooray!

Reviewed commit: 44ba95abc4

ℹ️ 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".

temporal-github-worker-1 Bot added agent/review-requested ecosystems/review-requested product/review-requested labels Oct 31, 2026

dd-octo-sts Bot added documentation integration/kafka_consumer labels Oct 31, 2026

dsavov force-pushed the dsavov/kafka-consumer-shard-groups branch from b1a5f53 to f506c7f Compare October 31, 2026 07:40

dsavov mentioned this pull request Oct 31, 2026

[clusterchecks] Add consumer group sharding for kafka_consumer DataDog/datadog-agent#59382

Open

dsavov marked this pull request as ready for review October 31, 2026 08:56

dsavov requested review from a team as code owners October 31, 2026 08:56

dd-octo-sts Bot added team/agent-integrations team/documentation labels Oct 31, 2026

datadog-datadog-us1-prod Bot reviewed Oct 31, 2026

View reviewed changes

chatgpt-codex-connector Bot reviewed Oct 31, 2026

View reviewed changes

dsavov force-pushed the dsavov/kafka-consumer-shard-groups branch from f506c7f to fbb9a23 Compare October 31, 2026 09:09

chatgpt-codex-connector Bot reviewed Oct 31, 2026

View reviewed changes

dsavov force-pushed the dsavov/kafka-consumer-shard-groups branch from fbb9a23 to 44ba95a Compare October 31, 2026 09:32



		def sharded_instance(kafka_instance, shard_index, shard_count=3, **options):
		kafka_instance.pop("consumer_groups", None)

Conversation

dsavov commented Oct 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What does this PR do?

Motivation

Checklist before requesting review

Uh oh!

dd-octo-sts Bot commented Oct 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Dispatcher tests: passed

Uh oh!

datadog-datadog-us1-prod Bot commented Oct 31, 2026 • edited by datadog-prod-us1-3 Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ Errors

ℹ️ Info

Uh oh!

dsavov commented Oct 31, 2026

Uh oh!

dsavov commented Oct 31, 2026

Uh oh!

chatgpt-codex-connector Bot commented Oct 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codex Review Summary

Uh oh!

datadog-datadog-us1-prod Bot left a comment

Choose a reason for hiding this comment

Uh oh!

datadog-datadog-us1-prod Bot Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

dsavov Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

chatgpt-codex-connector Bot Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

dsavov Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

dsavov commented Oct 31, 2026

Uh oh!

dsavov commented Oct 31, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

chatgpt-codex-connector Bot Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

dsavov Oct 31, 2026

Choose a reason for hiding this comment

Uh oh!

dsavov commented Oct 31, 2026

Uh oh!

chatgpt-codex-connector Bot commented Oct 31, 2026

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

dsavov commented Oct 31, 2026 •

edited

Loading

dd-octo-sts Bot commented Oct 31, 2026 •

edited

Loading

datadog-datadog-us1-prod Bot commented Oct 31, 2026 •

edited by datadog-prod-us1-3 Bot

Loading

chatgpt-codex-connector Bot commented Oct 31, 2026 •

edited

Loading