aglinxinyuan commented on code in PR #5700:
URL: https://github.com/apache/texera/pull/5700#discussion_r3565431366
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -87,19 +98,62 @@ def __init__(
target=self.data_processor.run, daemon=True,
name="data_processor_thread"
).start()
+ def _jump_to_loop_start(
+ self, executor: LoopEndOperator, coordinator_interface
+ ) -> None:
+ # The write address is setup config, keyed by the captured id. Fail
+ # loud BEFORE the jump RPC so a misconfigured loop does not rewind the
+ # schedule without a back-edge write.
+ uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ if not uri:
+ raise RuntimeError(
+ f"no loop-back state URI configured for LoopStart "
+ f"'{self._loop_start_id}' "
+ f"(have: {sorted(self.context.loop_start_state_uris)})"
+ )
+ coordinator_interface.jump_to_operator_region(
+ JumpToOperatorRegionRequest(OperatorIdentity(self._loop_start_id))
+ )
+ writer = DocumentFactory.create_document(uri, State.SCHEMA).writer("0")
+ # The back-edge fires only after the matching LoopEnd consumed at
+ # loop_counter == 0, so the next iteration's input starts at depth 0.
+ writer.put_one(executor.state.to_tuple(0))
+ writer.close()
+
def complete(self) -> None:
"""
Complete the DataProcessor, marking state to COMPLETED, and notify the
coordinator.
"""
# flush the buffered console prints
self._check_and_report_console_messages(force_flush=True)
- self.context.executor_manager.executor.close()
+ coordinator_interface = self._async_rpc_client.coordinator_stub()
+ executor = self.context.executor_manager.executor
+ if isinstance(executor, LoopEndOperator):
+ # condition() evaluates a user-supplied expression on this main
+ # loop thread. A UDF error on the data path is caught and
+ # reported (DataProcessor._executor_session); mirror that here so
+ # a bad condition (a typo, an undefined name) surfaces as an
+ # operator-facing error and pauses the worker, instead of killing
+ # the thread through run()'s @logger.catch(reraise=True).
+ try:
+ should_jump = executor.condition()
+ except Exception as err:
Review Comment:
Done — extracted the sequence into `Context.report_exception(err)`
(`logger.exception` → `set_exception_info` → error console message) and call it
from both `DataProcessor._executor_session` and `MainLoop.complete()`, so they
stay in sync. Also dropped the now-unused `sys` / `loguru` /
`create_error_console_message` imports from the two callers. (c940646)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
