At stop_consumer function pass additional config to AdminClient (#50)

robooo · Aug 1, 2024 · 4356830 · 4356830
1 parent 53fb139
commit 4356830
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/ConfluentKafkaLibrary/consumer.py b/src/ConfluentKafkaLibrary/consumer.py
@@ -34,7 +34,7 @@ def __init__(
                                                       server=server,
                                                       port=port,
                                                       **kwargs)
-
+        self.kwargs = kwargs
         if not isinstance(topics, list):
             topics = [topics]
         self.consumer.subscribe_topic(self.group_id, topics=topics)
@@ -65,7 +65,7 @@ def stop_consumer(self):
         self.join()
         self.consumer.unsubscribe(self.group_id)
         self.consumer.close_consumer(self.group_id)
-        admin_client = AdminClient({'bootstrap.servers': f'{self.server}:{self.port}'})
+        admin_client = AdminClient({'bootstrap.servers': f'{self.server}:{self.port}', **self.kwargs})
         response = admin_client.delete_consumer_groups([self.group_id], request_timeout=10)
         try:
             response[self.group_id].result()