fix: Fix SparkKafkaProcessor query_timeout parameter (#2789)

felixwang9817 · web-flow · commit a8d282d3e4f0 · 2022-06-13T13:18:09.000-07:00
Signed-off-by: Felix Wang &lt;wangfelix98@gmail.com&gt;
diff --git a/sdk/python/feast/infra/contrib/spark_kafka_processor.py b/sdk/python/feast/infra/contrib/spark_kafka_processor.py
@@ -17,6 +17,8 @@
 
 class SparkProcessorConfig(ProcessorConfig):
     spark_session: SparkSession
+    processing_time: str
+    query_timeout: int
 
 
 class SparkKafkaProcessor(StreamProcessor):
@@ -31,7 +33,7 @@ def __init__(
         config: ProcessorConfig,
         write_function: MethodType,
         processing_time: str = "30 seconds",
-        query_timeout: str = "15 seconds",
+        query_timeout: int = 15,
     ):
         if not isinstance(sfv.stream_source, KafkaSource):
             raise ValueError("data source is not kafka source")
diff --git a/sdk/python/feast/infra/contrib/stream_processor.py b/sdk/python/feast/infra/contrib/stream_processor.py
@@ -81,7 +81,7 @@ def get_stream_processor_object(
     if config.mode == "spark" and config.source == "kafka":
         stream_processor = STREAM_PROCESSOR_CLASS_FOR_TYPE[("spark", "kafka")]
         module_name, class_name = stream_processor.rsplit(".", 1)
-        cls = import_class(module_name, class_name, "Processor")
+        cls = import_class(module_name, class_name, "StreamProcessor")
         return cls(sfv=sfv, config=config, write_function=write_function,)
     else:
         raise ValueError("other processors besides spark-kafka not supported")
diff --git a/sdk/python/feast/stream_feature_view.py b/sdk/python/feast/stream_feature_view.py
@@ -132,10 +132,12 @@ def __eq__(self, other):
 
         if not super().__eq__(other):
             return False
+
         if not self.udf:
             return not other.udf
         if not other.udf:
             return False
+
         if (
             self.mode != other.mode
             or self.timestamp_field != other.timestamp_field
