fix: add the error to the global result queue (#315)

Signed-off-by: Vigith Maurice <vigith@gmail.com>

Author: vigith

packages/pynumaflow/pynumaflow/mapper/_servicer/_async_servicer.py

@@ -93,8 +93,10 @@ async def _process_inputs(
             # send an EOF to result queue to indicate that all tasks have completed
             await result_queue.put(STREAM_EOF)
 
-        except BaseException:
+        except BaseException as e:
             _LOGGER.critical("MapFn Error, re-raising the error", exc_info=True)
+            # Surface the error to the consumer; MapFn will handle and exit
+            await result_queue.put(e)
 
     async def _invoke_map(self, req: map_pb2.MapRequest, result_queue: NonBlockingIterator):
         """

packages/pynumaflow/pynumaflow/mapper/_servicer/_sync_servicer.py

@@ -92,8 +92,10 @@ def _process_requests(
             self.executor.shutdown(wait=True)
             # Indicate to the result queue that no more messages left to process
             result_queue.put(STREAM_EOF)
-        except BaseException:
+        except BaseException as e:
             _LOGGER.critical("MapFn Error, re-raising the error", exc_info=True)
+            # Surface the error to the consumer; MapFn will handle and exit
+            result_queue.put(e)
 
     def _invoke_map(
         self,

packages/pynumaflow/pynumaflow/sourcetransformer/servicer/_async_servicer.py

@@ -96,8 +96,10 @@ async def _process_inputs(
             # send an EOF to result queue to indicate that all tasks have completed
             await result_queue.put(STREAM_EOF)
 
-        except BaseException:
+        except BaseException as e:
             _LOGGER.critical("SourceTransformFnError Error, re-raising the error", exc_info=True)
+            # Surface the error to the consumer; SourceTransformFn will handle and exit
+            await result_queue.put(e)
 
     async def _invoke_transform(
         self, request: transform_pb2.SourceTransformRequest, result_queue: NonBlockingIterator

packages/pynumaflow/pynumaflow/sourcetransformer/servicer/_servicer.py

@@ -113,8 +113,10 @@ def _process_requests(
             self.executor.shutdown(wait=True)
             # Indicate to the result queue that no more messages left to process
             result_queue.put(STREAM_EOF)
-        except BaseException:
+        except BaseException as e:
             _LOGGER.critical("SourceTransformFnError, re-raising the error", exc_info=True)
+            # Surface the error to the consumer; SourceTransformFn will handle and exit
+            result_queue.put(e)
 
     def _invoke_transformer(
         self, context, request: transform_pb2.SourceTransformRequest, result_queue: SyncIterator