#138 - kafka.read - merge fixes

datayoga-io · Feb 18, 2024 · e5425ed · e5425ed
1 parent ddb1ecb
commit e5425ed
Show file tree

Hide file tree

Showing 6 changed files with 12 additions and 16 deletions.
diff --git a/core/pyproject.toml b/core/pyproject.toml
@@ -69,10 +69,9 @@ kafka = ["kafka-python", "confluent-kafka"]
 test = [
         "aiohttp",
         "cassandra-driver",
-        "confluent-kafka",
         "fastparquet",
         "ibm_db_sa",
-        "kafka-python",
+        "kafka",
         "mock",
         "oracledb",
         "psycopg2-binary",
@@ -87,7 +86,7 @@ test = [
         "redis",
         "requests-mock",
         "SQLAlchemy",
-        "testcontainers",
+        "testcontainers"
 ]
 
 [tool.poetry.urls]

diff --git a/core/src/datayoga_core/blocks/kafka/read/block.py b/core/src/datayoga_core/blocks/kafka/read/block.py
@@ -5,7 +5,7 @@
 from typing import AsyncGenerator, List, Optional
 
 import orjson
-from confluent_kafka import Consumer, KafkaError, TopicPartition
+from confluent_kafka import Consumer, KafkaError
 from datayoga_core import utils
 from datayoga_core.context import Context
 from datayoga_core.producer import Message
@@ -26,14 +26,12 @@ class Block(DyProducer, metaclass=ABCMeta):
 
     def init(self, context: Optional[Context] = None):
         logger.debug(f"Initializing {self.get_block_name()}")
-        connection_details = utils.get_connection_details(
-            self.properties["bootstrap_servers"], context)
+        connection_details = utils.get_connection_details(self.properties["bootstrap_servers"], context)
         logger.debug(f"Connection details: {json.dumps(connection_details)}")
         self.bootstrap_servers = connection_details.get("bootstrap_servers")
         self.group = self.properties.get("group")
         self.topic = self.properties["topic"]
-        self.seek_to_beginning = self.properties.get(
-            "seek_to_beginning", False)
+        self.seek_to_beginning = self.properties.get("seek_to_beginning", False)
         self.snapshot = self.properties.get("snapshot", False)
 
     async def produce(self) -> AsyncGenerator[List[Message], None]:

diff --git a/core/src/datayoga_core/resources/schemas/connections/kafka.schema.json b/core/src/datayoga_core/resources/schemas/connections/kafka.schema.json
@@ -9,7 +9,7 @@
       "const": "kafka"
     },
     "bootstrap_servers": {
-      "description": "Kafka Hosts",
+      "description": "Kafka Hosts List comma separated",
       "type": "string"
     }
   },
@@ -19,7 +19,7 @@
     {
       "kafka": {
         "type": "kafka",
-        "bootstrap_servers": ["localhost:9092"]
+        "bootstrap_servers": ["localhost:9092,localhost:9093"]
       }
     }
   ]

diff --git a/integration-tests/resources/jobs/tests/kafka_to_redis.yaml b/integration-tests/resources/jobs/tests/kafka_to_redis.yaml
@@ -2,8 +2,8 @@ input:
   uses: kafka.read
   with:
     bootstrap_servers: kafka
-    topic: "integration-tests"
-    group: "integration-tests"
+    topic: integration-tests
+    group: integration-tests
     snapshot: true
 steps:
   - uses: redis.write

diff --git a/integration-tests/resources/jobs/tests/kafka_to_stdout.yaml b/integration-tests/resources/jobs/tests/kafka_to_stdout.yaml
@@ -2,8 +2,8 @@ input:
   uses: kafka.read
   with:
     bootstrap_servers: kafka
-    topic: "integration-tests"
-    group: "integration-tests"
+    topic: integration-tests
+    group: integration-tests
     snapshot: true
 steps:
   - uses: std.write
diff --git a/integration-tests/test_kafka_to_redis.py b/integration-tests/test_kafka_to_redis.py
@@ -23,8 +23,7 @@ def test_kafka_to_redis():
             producer.flush()
             run_job("tests.kafka_to_redis")
 
-            redis_client = redis_utils.get_redis_client(
-                "localhost", redis_utils.REDIS_PORT)
+            redis_client = redis_utils.get_redis_client("localhost", redis_utils.REDIS_PORT)
 
             assert len(redis_client.keys()) == 2