[kafka_consumer] Purge stale broker_timestamps when offsets go backwards (DataDog#23409)

* kafka_consumer: purge stale broker_timestamps when offsets go backwards

When a topic's highwater offset decreases (retention wipe, topic
recreation, or offset reset), the cached (offset, timestamp) pairs with
offsets above the new highwater no longer correspond to real messages.
_get_interpolated_timestamp then extrapolates between those stale pairs
and fresh ones, pinning estimated_consumer_lag to a wall-clock value
equal to how long ago the reset happened (e.g. a steady ~18h lag
reading while kafka.consumer_lag shows only ~170 offsets).

Fix _add_broker_timestamps to:
  - drop cached entries with offsets above the new highwater
  - evict by oldest timestamp rather than smallest offset, so fresh
    post-reset entries are kept over poisonous stale ones.

* Add changelog fragment

Author: piochelepiotr

kafka_consumer/changelog.d/23409.fixed

@@ -0,0 +1 @@
+When a topic's highwater offset decreases (retention wipe, topic recreation, or offset reset), purge cached (offset, timestamp) pairs whose offset is above the new highwater and switch eviction to oldest-timestamp instead of smallest-offset. Previously, stale pre-reset entries poisoned interpolation and pinned `kafka.estimated_consumer_lag` to a wall-clock value equal to how long ago the reset happened.

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -259,10 +259,19 @@ def _mark_messages_retrieved(self, config_id):
     def _add_broker_timestamps(self, broker_timestamps, highwater_offsets):
         for (topic, partition), highwater_offset in highwater_offsets.items():
             timestamps = broker_timestamps["{}_{}".format(topic, partition)]
+            # If the highwater offset went backwards (topic recreated,
+            # retention wipe, or offset reset) any cached pair with a larger
+            # offset points to a now-nonexistent message and would poison
+            # interpolation. Drop those entries.
+            stale = [o for o in timestamps if o > highwater_offset]
+            for o in stale:
+                del timestamps[o]
             timestamps[highwater_offset] = time()
-            # If there's too many timestamps, we delete the oldest
+            # If there's too many timestamps, we delete the oldest one (by
+            # timestamp, not by offset — evicting by min offset would discard
+            # the fresh post-reset entries and keep poisonous stale ones).
             if len(timestamps) > self._max_timestamps:
-                del timestamps[min(timestamps)]
+                del timestamps[min(timestamps, key=timestamps.get)]
 
     def _save_broker_timestamps(self, broker_timestamps, persistent_cache_key):
         """Saves broker timestamps to persistent cache."""

kafka_consumer/tests/test_unit.py

@@ -512,6 +512,37 @@ def test_client_init(kafka_instance, check, dd_run_check):
     assert check.client.open_consumer.mock_calls == [mock.call("datadog-agent")]
 
 
+def test_add_broker_timestamps_purges_stale_offsets_on_reset(kafka_instance, check):
+    # When the highwater offset goes backwards (topic recreated / retention
+    # wipe / offset reset), cached (offset, timestamp) pairs with offsets
+    # above the new highwater are stale and must be purged — otherwise they
+    # poison interpolation and pin estimated_consumer_lag to a wall-clock
+    # offset equal to how long ago the reset happened.
+    check = check(kafka_instance)
+    broker_timestamps = {"topic1_0": {1_000_000: 100.0, 999_000: 99.0}}
+    check._add_broker_timestamps(broker_timestamps, {("topic1", 0): 170})
+
+    timestamps = broker_timestamps["topic1_0"]
+    assert 1_000_000 not in timestamps
+    assert 999_000 not in timestamps
+    assert 170 in timestamps
+
+
+def test_add_broker_timestamps_evicts_by_oldest_timestamp(kafka_instance, check):
+    # Eviction must drop the entry with the oldest timestamp, not the smallest
+    # offset. Evicting by min(offset) would discard fresh post-reset entries
+    # and keep poisonous ones.
+    kafka_instance['timestamp_history_size'] = 2
+    check = check(kafka_instance)
+    broker_timestamps = {"topic1_0": {500: 50.0, 400: 999.0}}
+    check._add_broker_timestamps(broker_timestamps, {("topic1", 0): 600})
+
+    timestamps = broker_timestamps["topic1_0"]
+    assert 500 not in timestamps  # oldest by timestamp
+    assert 400 in timestamps
+    assert 600 in timestamps
+
+
 def test_resolve_start_offsets():
     highwater_offsets = {
         ("topic1", 0): 100,