diff --git a/src/ConfluentKafkaLibrary/consumer.py b/src/ConfluentKafkaLibrary/consumer.py index 4196029..847626c 100644 --- 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()