Merge pull request #92 from miladmahmoodi/features/kafka-producer-args-support

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

Author: HosseinNejatiJavaremi

archipy/adapters/kafka/adapters.py

@@ -507,11 +507,12 @@ def _delivery_callback(error: KafkaError | None, message: Message) -> None:
             )
 
     @override
-    def produce(self, message: str | bytes) -> None:
+    def produce(self, message: str | bytes, key: str | None = None) -> None:
         """Produces a message to the configured topic.
 
         Args:
             message (str | bytes): The message to produce.
+            key (str | None, optional): The key for the message. Defaults to None.
 
         Raises:
             NetworkError: If there is a network error producing the message.
@@ -524,6 +525,7 @@ def produce(self, message: str | bytes) -> None:
                 topic=self._topic_name,
                 value=processed_message,
                 callback=self._delivery_callback,
+                key=key,
             )
         except Exception as e:
             self._handle_producer_exception(e, "produce")

archipy/adapters/kafka/ports.py

@@ -140,11 +140,12 @@ class KafkaProducerPort:
     """
 
     @abstractmethod
-    def produce(self, message: str | bytes) -> None:
+    def produce(self, message: str | bytes, key: str | None = None) -> None:
         """Produces a message to the configured topic.
 
         Args:
             message (str | bytes): The message to produce.
+            key (str | None, optional): The key for the message. Defaults to None.
 
         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)