kafka_consumer: don't report negative offsets as committed

kafka_consumer/changelog.d/24938.fixed

@@ -0,0 +1 @@
+Skip partitions with no committed offset instead of reporting a negative offset and inflated `kafka.consumer_lag`.

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -16,11 +16,7 @@
 from datadog_checks.kafka_consumer.cluster_metadata import ClusterMetadataCollector
 from datadog_checks.kafka_consumer.config import KafkaConfig
 from datadog_checks.kafka_consumer.connectors import KafkaConnectCollector
-from datadog_checks.kafka_consumer.constants import (
-    HIGH_WATERMARK,
-    KAFKA_INTERNAL_TOPICS,
-    OFFSET_INVALID,
-)
+from datadog_checks.kafka_consumer.constants import HIGH_WATERMARK, KAFKA_INTERNAL_TOPICS
 
 MAX_TIMESTAMPS = 1000
 
@@ -271,7 +267,14 @@ def get_consumer_offsets(self):
                 self.log.debug('RESULTS PARTITION: %s', partition)
                 self.log.debug('RESULTS OFFSET: %s', offset)
 
-                if offset == OFFSET_INVALID:
+                # A real committed offset is always a non-negative, monotonically increasing
+                # per-partition sequence number assigned by the broker. librdkafka reuses the
+                # same signed offset field for logical/sentinel values instead: OFFSET_BEGINNING
+                # (-2), OFFSET_END (-1), OFFSET_STORED (-1000), and OFFSET_INVALID (-1001) meaning
+                # "no committed offset". Checking `< 0` instead of `== OFFSET_INVALID` also covers
+                # any other negative sentinel librdkafka may return here.
+                # https://github.com/confluentinc/librdkafka/blob/master/src/rdkafka.h
+                if offset < 0:
                     continue
 
                 if (

kafka_consumer/tests/test_unit.py

@@ -277,6 +277,22 @@ def test_when_consumer_lag_less_than_zero_then_emit_event(check, kafka_instance,
     )
 
 
+def test_when_no_committed_offset_then_consumer_metrics_are_skipped(check, kafka_instance, dd_run_check, aggregator):
+    # Given: a partition with no committed offset, which librdkafka can surface as a negative
+    # logical offset (e.g. -1001 OFFSET_INVALID, or -2 OFFSET_BEGINNING) rather than a real offset.
+    mock_client = seed_mock_client()
+    mock_client.list_consumer_group_offsets.return_value = [("consumer_group1", [("topic1", "partition1", -2)])]
+    kafka_consumer_check = check(kafka_instance)
+    kafka_consumer_check.client = mock_client
+
+    # When
+    dd_run_check(kafka_consumer_check)
+
+    # Then: the partition is skipped rather than reporting a negative offset and inflated lag
+    aggregator.assert_metric("kafka.consumer_offset", count=0)
+    aggregator.assert_metric("kafka.consumer_lag", count=0)
+
+
 def test_when_collect_consumer_group_state_is_enabled(check, kafka_instance, dd_run_check, aggregator):
     mock_client = seed_mock_client()
     kafka_instance["collect_consumer_group_state"] = True