kafka_consumer: always fetch highwater offsets when cluster monitoring is enabled (#24149)

* kafka_consumer: always fetch highwater offsets when cluster monitoring is enabled

When enable_cluster_monitoring is true the consumer context count can easily
exceed the default max_partition_contexts (500), causing the check to skip
highwater offset collection entirely. This silently zeros out
kafka.topic.message_rate and stops kafka.broker_offset from being emitted,
because _collect_topic_metadata receives an empty highwater_offsets dict.

Bypass the context limit guard when cluster monitoring is active so that
highwater offsets are always fetched; the existing per-metric context caps in
report_highwater_offsets and report_consumer_offsets_and_lag still apply.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_consumer: add changelog entry for PR #24149

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_consumer: bypass context reporting limit when cluster monitoring is enabled

When enable_cluster_monitoring is true, report all consumer lag and highwater
offset metrics without capping at max_partition_contexts. Cluster monitoring
users need full cluster visibility by design, so capping metric reporting makes
no sense in that mode.

Uses float('inf') as the reporting limit, which works correctly with the
existing int comparisons in report_highwater_offsets and
report_consumer_offsets_and_lag (int == inf is False, int >= inf is False,
int < inf is True). Also suppresses the misleading "narrow your target" warning
in cluster monitoring mode.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: piochelepiotr

kafka_consumer/changelog.d/24149.fixed

@@ -0,0 +1 @@
+Fix kafka.broker_offset and kafka.topic.message_rate not being collected when enable_cluster_monitoring is true and the consumer context count exceeds max_partition_contexts.

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -67,7 +67,9 @@ def check(self, _):
         persistent_cache_key = "broker_timestamps_"
         consumer_contexts_count = self.count_consumer_contexts(consumer_offsets)
         try:
-            if consumer_contexts_count < self._context_limit:
+            # Cluster monitoring always requires highwater offsets (for topic.message_rate and other
+            # cluster metadata metrics), so bypass the consumer context limit in that case.
+            if consumer_contexts_count < self._context_limit or self.config._cluster_monitoring_enabled:
                 # Fetch highwater offsets
                 # Build partitions list or use all if configured
                 # If cluster monitoring is enabled, always fetch all broker highwater marks
@@ -100,7 +102,10 @@ def check(self, _):
             consumer_offsets,
             highwater_offsets,
         )
-        if total_contexts >= self._context_limit:
+        # When cluster monitoring is enabled, all offsets and lag metrics are reported regardless
+        # of context count so that the full cluster picture is always available.
+        reporting_limit = float('inf') if self.config._cluster_monitoring_enabled else self._context_limit
+        if total_contexts >= self._context_limit and not self.config._cluster_monitoring_enabled:
             self.warning(
                 """Discovered %s metric contexts - this exceeds the maximum number of %s contexts permitted by the
                 check. Please narrow your target by specifying in your kafka_consumer.yaml the consumer groups, topics
@@ -113,11 +118,11 @@ def check(self, _):
         if self.config._kafka_cluster_id_override:
             cluster_id = self.config._kafka_cluster_id_override
 
-        self.report_highwater_offsets(highwater_offsets, self._context_limit, cluster_id)
+        self.report_highwater_offsets(highwater_offsets, reporting_limit, cluster_id)
         self.report_consumer_offsets_and_lag(
             consumer_offsets,
             highwater_offsets,
-            self._context_limit - len(highwater_offsets),
+            reporting_limit - len(highwater_offsets),
             broker_timestamps,
             cluster_id,
         )