From 2f519be05a71731721c09d7e82fa7d27719ae39c Mon Sep 17 00:00:00 2001 From: khushal Date: Tue, 7 Jan 2025 18:52:08 +0530 Subject: [PATCH] (20250107) Added a few params to synchronous Kafka. --- utils_v2/queue/kafka/controllers/kafka.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/utils_v2/queue/kafka/controllers/kafka.py b/utils_v2/queue/kafka/controllers/kafka.py index 96fb130..5cd8770 100644 --- a/utils_v2/queue/kafka/controllers/kafka.py +++ b/utils_v2/queue/kafka/controllers/kafka.py @@ -138,7 +138,10 @@ class ProducerKafka: ca_file: str | None = None, cert_file: str | None = None, key_file: str | None = None, - client_id: str | int | None = None + client_id: str | int | None = None, + acks: int = 0, + retries: int = 1, + linger_ms: int = 0 ): """ @@ -156,7 +159,10 @@ class ProducerKafka: if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers] config = { "bootstrap.servers": ",".join(bootstrap_servers), - "security.protocol": security_protocol + "security.protocol": security_protocol, + "acks": acks, + "retries": retries, + "linger.ms": linger_ms } # Add the SSL security details: