[kafka_consumer] Fix consumer leak when offsets_for_times times out (DataDog#23241)

* [kafka_consumer] Fix consumer leak when offsets_for_times times out

Wrap Consumer usage in get_watermark_offsets() in try/finally to ensure
close_consumer() is always called, even when offsets_for_times() throws.

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

* Add changelog entry for kafka_consumer consumer leak fix

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

* Remove unnecessary debug log and keep fix minimal

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

* Restore pre-existing debug log

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/changelog.d/23241.fixed

@@ -0,0 +1 @@
+Fix consumer leak when offsets_for_times() times out, preventing a potential librdkafka crash

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -408,21 +408,22 @@ def get_watermark_offsets(self, partitions=None, mode=HIGH_WATERMARK):
 
         # Open consumer once for both cluster_id and offset fetching
         self.client.open_consumer(dd_consumer_group)
-        cluster_id, _ = self.client.consumer_get_cluster_id_and_list_topics(dd_consumer_group)
-
-        self.log.debug(
-            'Querying %s %s offsets',
-            len(topic_partitions_to_check),
-            'highwater' if mode == HIGH_WATERMARK else 'lowwater',
-        )
+        try:
+            cluster_id, _ = self.client.consumer_get_cluster_id_and_list_topics(dd_consumer_group)
 
-        result = {}
-        for topic, partition, offset in self.client.consumer_offsets_for_times(
-            partitions=topic_partitions_to_check, offset=mode
-        ):
-            result[(topic, partition)] = offset
+            self.log.debug(
+                'Querying %s %s offsets',
+                len(topic_partitions_to_check),
+                'highwater' if mode == HIGH_WATERMARK else 'lowwater',
+            )
 
-        self.client.close_consumer()
+            result = {}
+            for topic, partition, offset in self.client.consumer_offsets_for_times(
+                partitions=topic_partitions_to_check, offset=mode
+            ):
+                result[(topic, partition)] = offset
+        finally:
+            self.client.close_consumer()
 
         self.log.debug('Got %s %s offsets', len(result), 'highwater' if mode == HIGH_WATERMARK else 'lowwater')
         return result, cluster_id