(20250108) Made changes to the way synchronous Kafka can be configured and used.

utils_v2/queue/kafka/controllers/kafka.py

@@ -141,7 +141,8 @@ class ProducerKafka:
             client_id: str | int | None = None,
             acks: int = 0,
             retries: int = 1,
-            linger_ms: int = 0
+            linger_ms: int = 0,
+            misc_json: dict = None
     ):
 
         """
@@ -152,6 +153,11 @@ class ProducerKafka:
         :param cert_file: Needed for 'SSL' security protocol.
         :param key_file: Needed for 'SSL' security protocol.
         :param client_id: An identifier for one producer. Useful for debugging later.
+        :param acks: The no. of acknowledgements needed by the producer to consider the message as delivered. '0' means
+            fire-and-forget, 'all' or '-1' means that every in-sync replica must acknowledge. It could be any no.
+        :param retries: The no. of times the producer mut try to send the message in case of a failure.
+        :param linger_ms: The time in milliseconds to wait before sending a batch of messages to the broker.
+        :param misc_json: Any other custom configuration that is specific to the Confluent Kafka library.
         :return: The dictionary that needs to be passed as the 'conf' param when creating the producer.
         """
 
@@ -174,6 +180,11 @@ class ProducerKafka:
         # Add an identifier for debugging:
         if client_id: config["client.id"] = client_id
 
+        # Add the elements of the misc. JSON:
+        if misc_json:
+            for k, v in misc_json.items():
+                config[k] = v
+
         # Done here:
         return config
 
@@ -205,15 +216,16 @@ class ProducerKafka:
         if not self.__connected: self.connect()
         return self.__connected
 
-    def flush(self):
+    def flush(self, timeout: float = None):
 
         """
         Flushes the buffer entirely.
+        :param timeout: The timeout value in seconds.
         :return: None
         """
 
         start_time = time.time()
-        self.__producer.flush()
+        self.__producer.flush(timeout = timeout)
         message = f"Producer flushed in {time.time() - start_time:.5f} second(s)."
         self.__printer(message)
 
@@ -238,7 +250,8 @@ class ProducerKafka:
             key = None,
             topic = None,
             encoding = "utf-8",
-            callback = None
+            callback = None,
+            poll_immediately: bool = False
     ) -> bool:
 
         """
@@ -249,6 +262,8 @@ class ProducerKafka:
             be used by default.
         :param encoding: The encoding format.
         :param callback: The function to call for delivery reports.
+        :param poll_immediately: Whether, or not, you would like to run 'poll' method immediately after the produce
+            method.
         :return: True or False based on the success of the operation.
         """
 
@@ -265,6 +280,7 @@ class ProducerKafka:
                 key = key,
                 callback = callback
             )
+            if poll_immediately: self.__producer.poll(0)
 
             # Return with success if no exception occurred:
             return True
@@ -326,7 +342,8 @@ class ConsumerKafka:
             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,
+            misc_json: dict = None
     ):
 
         """
@@ -342,6 +359,7 @@ class ConsumerKafka:
         :param cert_file: Needed for 'SSL' security protocol.
         :param key_file: Needed for 'SSL' security protocol.
         :param client_id: An identifier for one producer. Useful for debugging later.
+        :param misc_json: Any other custom configuration that is specific to the Confluent Kafka library.
         :return: The dictionary that needs to be passed as the 'conf' param when creating the producer.
         """
 
@@ -365,6 +383,11 @@ class ConsumerKafka:
         config["group.id"] = group_id
         config["auto.offset.reset"] = auto_offset_reset
 
+        # Add the elements of the misc. JSON:
+        if misc_json:
+            for k, v in misc_json.items():
+                config[k] = v
+
         # Done here:
         return config
 

utils_v2/queue/kafka/models/message.py

@@ -116,7 +116,7 @@ class ConsumedKafkaMessage(BaseModel):
     )
 
     ts: AwareDatetime | None = Field(
-        description = "the time at which this message was sent to the queue",
+        description = "the time (utc) at which this message was sent to the queue",
         frozen = True
     )