sources: Ensure process was started before checking for error (#506)

If the first assignment a consumer get is empty the sources won't be
started. In that situation the app run loop will check for error in
sources and fail.

Author: ovv

quixstreams/sources/manager.py

@@ -117,6 +117,9 @@ def raise_for_error(self) -> None:
         Raise a `quixstreams.sources.manager.SourceException`
         if the child process was terminated with an exception.
         """
+        if not self.started:
+            return
+
         if super().is_alive():
             return
 

tests/test_quixstreams/test_sources/test_manager.py

@@ -99,7 +99,7 @@ def test_terminate_source(self):
         "when,exitcode", [("run", 0), ("cleanup", 0), ("stop", -9)]
     )
     @pytest.mark.parametrize("pickleable", [True, False])
-    def test_raise_for_error_run(self, when, exitcode, pickleable):
+    def test_raise_for_error(self, when, exitcode, pickleable):
         manager = SourceManager()
 
         finished = multiprocessing.Event()
@@ -110,6 +110,9 @@ def test_raise_for_error_run(self, when, exitcode, pickleable):
         source.configure(Topic("topic", None), None)
         process = manager.register(source)
 
+        # never raise when not started
+        manager.raise_for_error()
+
         manager.start_sources()
 
         finished.wait(1)