diff --git a/utils_v2/queue/kafka/controllers/kafka.py b/utils_v2/queue/kafka/controllers/kafka.py index 5cd8770..9647d52 100644 --- a/utils_v2/queue/kafka/controllers/kafka.py +++ b/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 diff --git a/utils_v2/queue/kafka/models/message.py b/utils_v2/queue/kafka/models/message.py index 465f0f8..80e7029 100644 --- a/utils_v2/queue/kafka/models/message.py +++ b/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 )