Emit metric when context limits exceeded

[chnn]

What does this PR do?

Emit a datadog.kafka_consumer.context_limit_budget metric in the Kafka Consumer integration that represents the number of contexts remaining before the context/partition limit is hit. The metric will be negative is the limit is exceeded. It is tagged by kafka_cluster_id.

Motivation

We use the Kafka Consumer metrics in various UIs in Data Streams Monitoring. When the context limit is exceeded, the check starts to drop data, which is not at all obvious to users looking at those metrics. I want to display a warning in the UI whenever the user is near or beyond their configured context limit along with instructions on how to raise it.

Review checklist (to be filled by reviewers)

• Feature or bugfix MUST have appropriate tests (unit, integration, e2e)
• Add the qa/skip-qa label if the PR doesn't need to be tested during QA.
• If you need to backport this PR to another branch, you can add the backport/<branch-name> label to the PR and it will automatically open a backport PR once this one is merged

[datadog-datadog-prod-us1-2]

⚠️ Tests

✨ Fix all issues with BitsAI or with Cursor

⚠️ Warnings

🧪 16 Tests failed

test_check_kafka from test_integration.py     (Fix with Cursor)

Some metrics are collected but not asserted:
Asserted Metrics:
	- kafka.broker_offset
	- kafka.consumer_lag
	- kafka.consumer_offset
Found metrics that are not asserted:
	- kafka.datadog.kafka_consumer.context_limit_budget

test_check_kafka_metrics_limit from test_integration.py     (Fix with Cursor)

assert 2 == 1
 +  where 2 = len(defaultdict(<class 'list'>, {'kafka.datadog.kafka_consumer.context_limit_budget': [MetricStub(name='kafka.datadog.kafka_consumer.context_limit_budget', type=0, value=0.0, tags=['kafka_cluster_id:'], hostname='', device=None, flush_first_value=False)], 'kafka.consumer_offset': [MetricStub(name='kafka.consumer_offset', type=0, value=75.0, tags=['topic:marvel', 'partition:0', 'consumer_group:my_consumer', 'kafka_cluster_id:', 'optional:tag1'], hostname='', device=None, flush_first_value=False)]}))
 +    where defaultdict(<class 'list'>, {'kafka.datadog.kafka_consumer.context_limit_budget': [MetricStub(name='kafka.datadog.kafka_consumer.context_limit_budget', type=0, value=0.0, tags=['kafka_cluster_id:'], hostname='', device=None, flush_first_value=False)], 'kafka.consumer_offset': [MetricStub(name='kafka.consumer_offset', type=0, value=75.0, tags=['topic:marvel', 'partition:0', 'consumer_group:my_consumer', 'kafka_cluster_id:', 'optional:tag1'], hostname='', device=None, flush_first_value=False)]}) = <datadog_checks.base.stubs.aggregator.AggregatorStub object at 0x7fa0c7aa3380>._metrics

test_config[Empty consumer_groups and monitor_unlisted_consumer_groups true] from test_integration.py     (Fix with Cursor)

Metadata assertion errors using metadata.csv:
	- Expect \`kafka.datadog.kafka_consumer.context_limit_budget\` to be in metadata.csv.

View all

ℹ️ Info

No other issues found (see more)

❄️ No new flaky tests detected

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 4f6bceb | Docs | Datadog PR Page | Was this helpful? React with 👍/👎 or give us feedback!}