feat: add *args, **kwargs to support to Kafka producer

Author: miladmahmoodi

archipy/adapters/kafka/adapters.py

@@ -507,11 +507,13 @@ def _delivery_callback(error: KafkaError | None, message: Message) -> None:
             )
 
     @override
-    def produce(self, message: str | bytes) -> None:
+    def produce(self, message: str | bytes, *args, **kwargs) -> None:
         """Produces a message to the configured topic.
 
         Args:
             message (str | bytes): The message to produce.
+            *args: Additional positional arguments passed to the underlying producer.
+            **kwargs: Additional keyword arguments passed to the underlying producer.
 
         Raises:
             NetworkError: If there is a network error producing the message.
@@ -521,9 +523,11 @@ def produce(self, message: str | bytes) -> None:
         try:
             processed_message = self._pre_process_message(message)
             self._adapter.produce(
+                *args,
                 topic=self._topic_name,
                 value=processed_message,
                 callback=self._delivery_callback,
+                **kwargs,
             )
         except Exception as e:
             self._handle_producer_exception(e, "produce")

archipy/adapters/kafka/ports.py

@@ -140,11 +140,13 @@ class KafkaProducerPort:
     """
 
     @abstractmethod
-    def produce(self, message: str | bytes) -> None:
+    def produce(self, message: str | bytes, *args, **kwargs) -> None:
         """Produces a message to the configured topic.
 
         Args:
             message (str | bytes): The message to produce.
+            *args: Additional positional arguments passed to the underlying producer.
+            **kwargs: Additional keyword arguments passed to the underlying producer.
 
         Raises:
             NotImplementedError: If the method is not implemented by the concrete class.

features/kafka_adapters.feature

@@ -24,6 +24,12 @@ Feature: Kafka Adapter Operations Testing
     When I validate the producer health
     Then the producer health check should pass
 
+  Scenario: Produce message with additional parameters
+    Given a Kafka producer for topic "test-topic"
+    And a Kafka consumer subscribed to topic "test-topic" with group "test-group"
+    When I produce one message "Hello Kafka with key" with key "test-key" to topic "test-topic"
+    Then the consumer should receive message "Hello Kafka with key" from topic "test-topic"
+
   Scenario: Delete a topic
     Given a topic named "test-topic-deletable" exists
     When I delete the topic "test-topic-deletable"

features/steps/kafka_adapter_steps.py

@@ -160,6 +160,18 @@ def step_produce_message(context, message, topic_name):
         raise e
 
 
+@when('I produce one message "{message}" with key "{key}" to topic "{topic_name}"')
+def step_produce_message_with_key(context, message, key, topic_name):
+    adapter = get_kafka_producer_adapter(context, topic_name)
+    try:
+        adapter.produce(message, key=key)
+        adapter.flush(timeout=1)
+        context.logger.info(f"Produced message '{message}' to '{topic_name}' with key '{key}'")
+    except Exception as e:
+        context.logger.exception(f"Failed to produce message with key: {str(e)}")
+        raise e
+
+
 @when("I validate the producer health")
 def step_validate_health(context):
     scenario_context = get_current_scenario_context(context)