Merge pull request #439 from nodestream-proj/kafka-consumer-factory

Allowing additional params to be sent to KafkaConsumerFactory

Author: rreddy2

nodestream/pipeline/extractors/streams/kafka.py

@@ -17,12 +17,14 @@ def build_consumer_config(
         security_protocol: str,
         group_id: str,
         bootstrap_servers: List[str],
+        **kwargs: Any,
     ):
         return {
             "bootstrap.servers": ",".join(bootstrap_servers),
             "auto.offset.reset": offset_reset,
             "security.protocol": security_protocol,
             "group.id": group_id,
+            **kwargs,
         }
 
     @classmethod
@@ -33,10 +35,15 @@ def new(
         bootstrap_servers: List[str],
         offset_reset: str,
         security_protocol: str,
+        **kwargs: Any,
     ):
         consumer = Consumer(
             cls.build_consumer_config(
-                offset_reset, security_protocol, group_id, bootstrap_servers
+                offset_reset,
+                security_protocol,
+                group_id,
+                bootstrap_servers,
+                **kwargs,
             )
         )
         consumer.subscribe([topic_name])

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nodestream"
-version = "0.14.15"
+version = "0.14.16"
 description = "A Fast, Declarative ETL for Graph Databases."
 license = "GPL-3.0-only"
 authors = [