(texera) 01/01: feat(pyamber): report batch processing errors (#8298)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8298-ec3a9dd3ca9c361cbd7f339d8bc78669024e84e8
in repository https://gitbox.apache.org/repos/asf/texera.git

commit b3fef994f65af05cca72ff8b55e586f5170f82d2
Author: carloea2 <ca...@uci.edu>
AuthorDate: Wed Sep 9 07:22:06 2026 +0000

    feat(pyamber): report batch processing errors (#8298)
    
    ### What changes were proposed in this PR?
    
    Report per-row MainLoop failures through the worker exception path,
    enter exception pause, and stop consuming the affected batch.
    
    Before: the worker logged the failure, continued later rows, and could
    finish with missing output.
    
    After: the coordinator receives the error and the batch stops at the
    failed row.
    
    ### Any related issues, documentation, discussions?
    
    Closes #8297
    
    ### How was this PR tested?
    
    Added a regression covering the failure report, exception pause, and
    stopped batch consumption.
    
    ```powershell
    $env:PYTHONDONTWRITEBYTECODE='1'; C:\Users\carlo\texera\texera\.venv312\Scripts\python.exe -c "import sys,pytest; sys.path[:0]=[r'C:\Users\carlo\texera\texera-worktrees\investigate-bug78\amber\src\main\python',r'C:\Users\carlo\texera\texera\amber\src\main\python']; raise SystemExit(pytest.main([r'amber/src/test/python/core/runnables/test_main_loop.py::TestMainLoop::test_failing_element_reports_exception_and_stops_batch','-p','no:cacheprovider','-q']))"
    C:\Users\carlo\texera\texera\.venv312\Scripts\ruff.exe check amber/src/main/python/core/runnables/main_loop.py amber/src/test/python/core/runnables/test_main_loop.py
    C:\Users\carlo\texera\texera\.venv312\Scripts\ruff.exe format --check amber/src/main/python/core/runnables/main_loop.py amber/src/test/python/core/runnables/test_main_loop.py
    ```
    
    The focused regression passed. The full MainLoop test file was also run:
    36 tests passed and 8 pre-existing integration cases failed during
    executor initialization because the shared generated protocol files do
    not yet contain `loop_start_port_uris`.
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Codex
---
 amber/src/main/python/core/runnables/main_loop.py  |  4 +-
 .../test/python/core/runnables/test_main_loop.py   | 68 ++++++----------------
 2 files changed, 20 insertions(+), 52 deletions(-)

diff --git a/amber/src/main/python/core/runnables/main_loop.py b/amber/src/main/python/core/runnables/main_loop.py
index 108078ba4f..f22ffde904 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -811,7 +811,9 @@ class MainLoop(StoppableQueueBlockingRunnable):
                     self._process_state_frame,
                 )
             except Exception as err:
-                logger.exception(err)
+                self.context.report_exception(err)
+                self._check_exception()
+                return
 
     def _send_console_message(self, console_message: ConsoleMessage):
         self._async_rpc_client.coordinator_stub().console_message_triggered(
diff --git a/amber/src/test/python/core/runnables/test_main_loop.py b/amber/src/test/python/core/runnables/test_main_loop.py
index c5a566ada9..ce88db6614 100644
--- a/amber/src/test/python/core/runnables/test_main_loop.py
+++ b/amber/src/test/python/core/runnables/test_main_loop.py
@@ -3738,20 +3738,19 @@ class TestMainLoop:
         assert len(console_msgs) == 2
 
     @pytest.mark.timeout(5)
-    def test_a_failing_element_does_not_abandon_the_rest_of_the_batch(
+    def test_failing_element_reports_exception_and_stops_batch(
         self, main_loop, monkeypatch, mock_raw_schema
     ):
-        # The batch iterator is the only handle on the elements still to come,
-        # so letting a failure on one element escape the loop would silently
-        # drop every element behind it. _process_data_element's per-element
-        # backstop keeps iterating instead.
-        #
-        # Deliberately NOT asserted: that nothing is reported to the
-        # coordinator. The backstop only logs, so a runtime-level per-element
-        # failure never reaches Context.report_exception and the workflow can
-        # report success on a short result -- arguably a silent-wrong-result
-        # defect. Pinning that half would cement it, so this test asserts only
-        # that iteration continues and that nothing propagates.
+        console_msgs = []
+        pauses = []
+        monkeypatch.setattr(
+            main_loop, "_send_console_message", lambda msg: console_msgs.append(msg)
+        )
+        monkeypatch.setattr(
+            main_loop.context.pause_manager,
+            "pause",
+            lambda reason: pauses.append(reason),
+        )
         schema = Schema(raw_schema=mock_raw_schema)
         port_0 = PortIdentity(0, internal=False)
         channel = self._register_input_port(main_loop, schema, port_0, "sender")
@@ -3783,44 +3782,11 @@ class TestMainLoop:
             main_loop.context.statistics_manager, "increase_input_statistics", _boom
         )
 
-        # Because the coordinator report is deliberately not asserted (above),
-        # the log line is the swallow's ONLY remaining trace -- so pin it.
-        # `except Exception: pass` is a strictly worse regression than the
-        # defect described above (a short result with no evidence anywhere
-        # rather than a stack trace in the worker log) and is not a defensible
-        # production change, so this is a gap rather than a bug to cement.
-        # The proxy delegates every other level to the real logger so the
-        # module's debug/info calls keep working.
-        from core.runnables import main_loop as main_loop_module
-
-        class _RecordingLogger:
-            def __init__(self, delegate):
-                self.exceptions = []
-                self._delegate = delegate
-
-            def exception(self, err):
-                self.exceptions.append(err)
-
-            def __getattr__(self, name):
-                return getattr(self._delegate, name)
-
-        recorder = _RecordingLogger(main_loop_module.logger)
-        monkeypatch.setattr(main_loop_module, "logger", recorder)
-
-        # Must not raise.
         main_loop._process_data_element(element)
 
-        assert attempted == [0, 1], (
-            "a failure on one element must not abandon the rest of the batch; "
-            f"attempted: {attempted}"
-        )
-        assert [type(err) for err in recorder.exceptions] == [
-            RuntimeError,
-            RuntimeError,
-        ], (
-            "every swallowed per-element failure must leave a log trace; "
-            f"logged: {recorder.exceptions}"
-        )
-        assert all(
-            "statistics backend unavailable" in str(err) for err in recorder.exceptions
-        ), f"the logged trace must carry the real error; logged: {recorder.exceptions}"
+        assert attempted == [0]
+        assert main_loop.context.exception_manager.has_exception()
+        assert pauses == [PauseType.EXCEPTION_PAUSE]
+        error_msgs = [m for m in console_msgs if m.msg_type == ConsoleMessageType.ERROR]
+        assert len(error_msgs) == 1
+        assert "statistics backend unavailable" in error_msgs[0].title