kafka_consumer: reuse AdminClient and Consumer connections across runs (DataDog#24552)

* kafka_consumer: reuse AdminClient and Consumer across runs

Recreating the librdkafka AdminClient (close_admin_client defaulted to true) and
opening/closing a Consumer on every run tore down and respawned librdkafka's
per-broker threads each collection. On high-core hosts that spreads allocations
across many glibc arenas whose freed memory is never reused or returned, so agent
RSS climbs unbounded. Keep both clients alive (default close_admin_client to
false, make open_consumer idempotent, and stop closing the consumer used for
cluster-id detection) so the same threads and arenas are reused each run.

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

* Add changelog entry for connection reuse

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

* Drop inline rationale comments (kept in PR/changelog)

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

* Guard consumer reuse under close_admin_client flag

close_admin_client now controls reuse of both clients: when false (default) the
AdminClient and Consumer are both reused across runs; when true, both are closed
each run (previous behavior).

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

* Drop explanatory comment on consumer close guard

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

* Keep close_admin_client default unchanged (true)

Reuse of both clients is opt-in via close_admin_client: false; the default
behavior (closing clients each run) is unchanged.

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

* Reword changelog: reuse is opt-in via close_admin_client

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

---------

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

Author: piochelepiotr

kafka_consumer/changelog.d/24552.fixed

@@ -0,0 +1 @@
+Allow reusing the Kafka AdminClient and Consumer across check runs (via close_admin_client: false) to avoid unbounded agent memory growth from librdkafka thread churn.

kafka_consumer/datadog_checks/kafka_consumer/client.py

@@ -42,6 +42,9 @@ def kafka_client(self):
         return self._kafka_client
 
     def open_consumer(self, consumer_group):
+        if self._consumer is not None:
+            return
+
         config = {
             "bootstrap.servers": self.config._kafka_connect_str,
             "group.id": consumer_group,
@@ -55,8 +58,11 @@ def open_consumer(self, consumer_group):
         self.log.debug("Consumer instance %s created for group %s", self._consumer, consumer_group)
 
     def close_consumer(self):
+        if self._consumer is None:
+            return
         self.log.debug("Closing consumer instance %s", self._consumer)
         self._consumer.close()
+        self._consumer = None
 
     def __get_authentication_config(self):
         config = {

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -543,7 +543,8 @@ def get_highwater_offsets(self, partitions=None):
             ):
                 result[(topic, partition)] = offset
         finally:
-            self.client.close_consumer()
+            if self.config._close_admin_client:
+                self.client.close_consumer()
 
         self.log.debug('Got %s highwater offsets', len(result))
         return result, cluster_id