Optimize Kafka config queries (DataDog#22721)

* Optimize Kafka config queries

* add changelog

* report schema versions correctly

* Address PR comments

* fmt

* fix tests

* Point to Data Streams from readme

* Mention Data Streams in enable_cluster_monitoring

* Add size limit to schema latest version cache to prevent unbounded disk growth

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Sync conf.yaml.example after rebase

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Fix misleading docstring in _save_latest_version_cache

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: piochelepiotr

kafka_consumer/README.md

@@ -60,16 +60,16 @@ instances:
 
 ### Cluster Monitoring (Preview)
 
-In addition to consumer lag metrics, this integration can collect comprehensive cluster metadata when `enable_cluster_monitoring` is enabled:
+When `enable_cluster_monitoring` is enabled, the integration collects cluster-wide metrics for [Data Streams Monitoring][18] in addition to consumer lag:
 
-- **Broker information**: Configuration and health metrics
-- **Topic and partition details**: Sizes, offsets, replication status
-- **Consumer group metadata**: Member details and group state
-- **Schema registry**: Schema information (if `schema_registry_url` is provided)
+- **Brokers**: Configuration and health metrics
+- **Topics and partitions**: Sizes, offsets, and replication status
+- **Consumer groups**: Member details and group state
+- **Schema registry**: Schema metadata (requires `schema_registry_url`)
 
-All cluster monitoring metrics are tagged with `kafka_cluster_id` for easy filtering.
+#### Batched collection
 
-**Note**: This feature is in Preview and may increase Agent resource consumption on large clusters. The integration caches configuration and schema events to reduce volume.
+Broker configurations, topic configurations, and schema registry version checks are collected in batches across multiple agent runs rather than all at once. This reduces load on large Kafka clusters but means that not all metrics are emitted in every check run. On a cluster with many brokers, topics, or schema subjects, the integration spreads the work over successive runs so that each run stays fast and does not overload the cluster.
 
 Example configuration:
 ```yaml

kafka_consumer/assets/configuration/spec.yaml

@@ -310,7 +310,8 @@ files:
           type: string
       - name: enable_cluster_monitoring
         description: |
-          (PREVIEW) Enable cluster monitoring to collect broker, topic, partition, consumer group,
+          (PREVIEW) Enable cluster monitoring to collect cluster-wide metrics for Data Streams
+          Monitoring. Collects broker, topic, partition, consumer group,
           and schema registry metadata. Events are cached and tagged with kafka_cluster_id.
           May increase resource consumption on large clusters.
 
@@ -323,6 +324,18 @@ files:
           type: boolean
           example: false
           display_default: false
+      - name: kafka_configs_refresh_interval
+        description: |
+          How often (in seconds) to re-fetch broker configurations, topic configurations,
+          and schema registry version checks from Kafka. A longer interval reduces load on
+          the Kafka cluster and Schema Registry. On clusters with many topics or schemas,
+          the check fetches in batches and gracefully degrades by re-fetching less often if
+          it cannot keep up within this interval.
+        fleet_configurable: true
+        value:
+          type: integer
+          example: 180
+          display_default: 180
       - name: schema_registry_url
         description: |
           Schema Registry URL. When set with enable_cluster_monitoring, collects schema information.

kafka_consumer/changelog.d/22721.changed

@@ -0,0 +1 @@
+Broker configurations, topic configurations, and schema registry version checks are now collected in batches across multiple agent runs instead of all at once. This reduces load on large clusters but means that not all metrics are emitted in every check run. The batch sizes and refresh interval are controlled by the `kafka_configs_refresh_interval` configuration option.