Fix kafka reader. (#310)

quintoandar · Apr 14, 2021 · 0f31164 · 0f31164
1 parent 43392f4
commit 0f31164
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/butterfree/extract/readers/kafka_reader.py b/butterfree/extract/readers/kafka_reader.py
@@ -174,7 +174,7 @@ def consume(self, client: SparkClient) -> DataFrame:
         """
         # read using client and cast key and value columns from binary to string
         raw_df = (
-            client.read(format="kafka", options=self.options, stream=self.stream)
+            client.read(format="kafka", stream=self.stream, **self.options)
             .withColumn("key", col("key").cast("string"))
             .withColumn("value", col("value").cast("string"))
         )

diff --git a/setup.py b/setup.py
@@ -1,7 +1,7 @@
 from setuptools import find_packages, setup
 
 __package_name__ = "butterfree"
-__version__ = "1.2.0.dev8"
+__version__ = "1.2.0.dev9"
 __repository_url__ = "https://github.com/quintoandar/butterfree"
 
 with open("requirements.txt") as f:

diff --git a/tests/unit/butterfree/extract/readers/test_kafka_reader.py b/tests/unit/butterfree/extract/readers/test_kafka_reader.py
@@ -99,7 +99,7 @@ def test_consume(
 
         # assert
         spark_client.read.assert_called_once_with(
-            format="kafka", options=options, stream=kafka_reader.stream
+            format="kafka", stream=kafka_reader.stream, **options
         )
         assert_dataframe_equality(target_df, output_df)