aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3671354638
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -97,21 +99,40 @@ 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. Anything raised here (a missing
- # URI, or a failed state write after the jump) is reported by
- # complete()'s guard as an operator-facing error.
- uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ def _loop_start_base_uri(self) -> str:
+ # The loop's bookkeeping base URI is setup config, keyed by the
+ # captured id (see InitializeExecutorRequest.loopStartPortUris). Fail
+ # loud on a missing entry: anything raised here is reported by the
+ # caller's guard as an operator-facing error.
+ uri = self.context.loop_start_port_uris.get(self._loop_start_id)
if not uri:
raise RuntimeError(
- f"no loop-back state URI configured for LoopStart "
+ f"no loop bookkeeping URI configured for LoopStart "
f"'{self._loop_start_id}' "
- f"(have: {sorted(self.context.loop_start_state_uris)})"
+ f"(have: {sorted(self.context.loop_start_port_uris)})"
)
+ return uri
+
+ def _read_loop_input_table(self) -> Table:
+ # The loop's input table is the Loop Start's input-port
+ # materialization -- data that already exists for the whole loop
+ # (Loop Start re-reads it every iteration; the back-edge truncates
+ # only the state doc at the same base URI, never the result doc).
+ # Reading it here at consume time means the table never has to ride
+ # inside the State content through the loop body.
+ result_uri = VFSURIFactory.result_uri(self._loop_start_base_uri())
+ document, _ = DocumentFactory.open_document(result_uri)
Review Comment:
Thanks — that was a real problem on the commit you looked at, and the
diagnosis was right: the loop table read was racing this worker's own
materialization reader. Reading a second iceberg/S3 doc from that thread while
the reader iterates a lazily-pinned snapshot that region re-execution drops and
recreates makes the reader fail with S3 `Access Denied` (MinIO's answer for a
deleted key).
Fixed in `822078ca7f` — the read is deferred to the EndChannel path, where
this worker's reader has already finished, so the two can never overlap. That
landed about half an hour after your comment, so the run you saw predates it.
On the current head:
| leg | before | now |
|---|---|---|
| amber-integration (ubuntu) | cancelled at 20m | **passed, 9m04s** |
| amber-integration (macos) | cancelled at 20m | fails: 40 succeeded, 1
failed |
`Access Denied` no longer appears anywhere in either log (0 occurrences).
The remaining macos failure looks like a different, already-tracked issue
rather than this refactor: it fails with `RuntimeError: worker still has
unprocessed messages` followed by `Failed to terminate region N on attempt 1 of
150` — i.e. #6891 (*Flaky loop e2e tests: EndWorker fails on in-flight RPC acks
during region termination*), whose fix is in flight as #6960.
I'm leaving this thread open rather than resolving it, since macos is still
red — happy to rebase on #6960 once it lands to confirm, if that's the cleanest
way to show it green.
--
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]
