[Backport 7.81.x] Kafka consumer broker timestamps and earliest offsets caching

[piochelepiotr]

What does this PR do?

Backports six kafka_consumer changes to the 7.81.x branch:

• [Backport 7.81.x] kafka_consumer: store broker_timestamps as marshal instead of json #24472: Persist the DSM broker timestamps cache using marshal instead of JSON, to reduce cluster-check-runner memory churn.
• kafka_consumer: cap total broker_timestamps history by a per-cluster memory budget #24508: Cap the retained broker_timestamps history per cluster by a per-partition budget derived from MAX_TIMESTAMP_ENTRIES, so memory use no longer grows unbounded with partition count.
• kafka_consumer: cache earliest offsets across collection intervals #24515: Cache earliest (log-start) offsets per broker-config-derived TTL instead of fetching them on every check run, and refetch from the broker (without extending the TTL) if a fresh cache doesn't cover every requested partition.
• kafka_consumer: reuse AdminClient and Consumer connections across runs #24552: Reuse the librdkafka AdminClient and Consumer across runs (default close_admin_client to false, idempotent open_consumer, stop closing the cluster-id consumer) so per-broker threads and their glibc arenas persist and their memory is recycled in place instead of stranded each run.
• kafka_consumer: return freed memory to the OS with malloc_trim after each run #24553: Call malloc_trim(0) after each run (glibc/Linux only) to return per-arena free memory to the OS.
• kafka_consumer: keep broker_timestamps in memory and persist every 5 minutes #24554: Keep the DSM broker_timestamps cache in memory across runs and persist it to disk at most every 5 minutes, removing the per-run marshal load/dump churn that scaled with cache size.

The #24508 and #24515 changes were adapted during the backport: #24508 was built on master's newer bulk-pruning _add_broker_timestamps (with prune_floors/_visvalingam_whyatt), so it was reimplemented on top of the 7.81.x branch's older single-oldest-eviction logic while preserving the same budget-scaling behavior. #24515's public fetch_earliest_offsets method was kept as the 7.81.x branch's existing private _fetch_earliest_offsets name to avoid an unrelated naming change.

Motivation

Bring memory-usage and caching improvements already merged/in review on master to the 7.81.x release branch.

Review checklist (to be filled by reviewers)

• Feature or bugfix MUST have appropriate tests (unit, integration, e2e)
• Add qa/required if this PR needs QA validation, or qa/skip-qa if it does not. Exactly one of the two is required.
• 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]

 

🎉 All green!

🧪 All tests passed
❄️ No new flaky tests detected

🎯 Code Coverage (details)
• Patch Coverage: 85.53%
• Overall Coverage: 88.46%

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 50e77c7 | Docs | Datadog PR Page | Give us feedback!}

[dd-octo-sts]

Validation Report

All 21 validations passed.

Show details

Validation	Description	Status
agent-reqs	Verify check versions match the Agent requirements file	✅
ci	Validate CI configuration and code coverage settings	✅
codeowners	Validate every integration has a CODEOWNERS entry	✅
config	Validate default configuration files against spec.yaml	✅
dep	Verify dependency pins are consistent and Agent-compatible	✅
http	Validate integrations use the HTTP wrapper correctly	✅
imports	Validate check imports do not use deprecated modules	✅
integration-style	Validate check code style conventions	✅
jmx-metrics	Validate JMX metrics definition files and config	✅
labeler	Validate PR labeler config matches integration directories	✅
legacy-signature	Validate no integration uses the legacy Agent check signature	✅
license-headers	Validate Python files have proper license headers	✅
licenses	Validate third-party license attribution list	✅
metadata	Validate metadata.csv metric definitions	✅
models	Validate configuration data models match spec.yaml	✅
openmetrics	Validate OpenMetrics integrations disable the metric limit	✅
package	Validate Python package metadata and naming	✅
qa-label	Validate the pull request declares whether it needs QA for the next Agent release	✅
readmes	Validate README files have required sections	✅
saved-views	Validate saved view JSON file structure and fields	✅
version	Validate version consistency between package and changelog	✅

View full run