py: run spwaned listeners as daemons (#1834)

abhizer · web-flow · commit 5f6f3bf38cdb · 2024-06-04T22:18:57.000+05:45
The CallbackRunner, which runs in a separate thread now runs as a
daemon. This means that the spawned thread will stop executing when the
main thread ends.

Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/python/feldera/_callback_runner.py b/python/feldera/_callback_runner.py
@@ -23,6 +23,7 @@ def __init__(
             queue: Optional[Queue],
     ):
         super().__init__()
+        self.daemon = True
         self.client: FelderaClient = client
         self.pipeline_name: str = pipeline_name
         self.view_name: str = view_name
diff --git a/python/tests/test_wireframes.py b/python/tests/test_wireframes.py
@@ -119,6 +119,18 @@ def test_df_without_columns(self):
         with self.assertRaises(ValueError):
             sql.connect_source_pandas(TBL_NAME, df)
 
+    def test_sql_error(self):
+        sql = SQLContext('sql_error', TEST_CLIENT).get_or_create()
+        TBL_NAME = "student"
+        df = pd.DataFrame([(1, "a"), (2, "b"), (3, "c")], columns=["id", "name"])
+        sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"}))
+        sql.register_view("s", f"SELECT FROM blah")
+        sql.connect_source_pandas(TBL_NAME, df)
+        _ = sql.listen("s")
+
+        with self.assertRaises(Exception):
+            sql.run_to_completion()
+
     def test_kafka(self):
         from kafka import KafkaProducer
         from kafka.admin import KafkaAdminClient, NewTopic
