eugenegujing opened a new pull request, #6724: URL: https://github.com/apache/texera/pull/6724
### What changes were proposed in this PR? In the Python worker, pause and backpressure work by disabling the data sub-queues of `InternalQueue`. However, `disable_data` only disables the sub-queues that exist when it is called, while sub-queues are created lazily on a channel's first `put` and start enabled. A data channel whose first message arrives during a pause or backpressure window therefore comes up enabled and is not covered by the disable at all. This is a bug because the worker can then keep processing data while reporting PAUSED, and if the leaked `DataElement` is dequeued inside the pause wait-loop in `main_loop.py`, the control-only pampy `match` raises an uncaught `MatchError` that silently kills the DP thread while the heartbeat thread stays alive, so the execution hangs forever with no error reported. Under backpressure, the leaked channel keeps feeding the congested downstream, defeating flow control. This PR fixes `InternalQueue.put`: when it registers a new data channel while `_queue_state` is non-empty, the new sub-queue now starts disabled. The registration is done under the same lock used by `disable_data`/`enable_data` (with a double-check to avoid duplicate registration), and the sub-queue is disabled before its first element is enqueued, so the element never becomes dequeuable during the disable window. Control channels are never disabled, and `enable_data` needs no change because it iterates the channel set at release time, which by then includes channels registered mid-pause. Cleanup in the same method: the unreachable `InternalMarker` entry in the isinstance tuple is removed (`InternalMarker` does not subclass `InternalQueueElement`, so it can never reach that branch), and the two identical dispatch branches are merged. ### Any related issues, documentation, discussions? Fixes #6723. The regression tests extend the `InternalQueue` spec added by #6444. ### How was this PR tested? Added 8 regression tests to `amber/src/test/python/core/models/test_internal_queue.py`: the late-channel leak (fails without the fix), release via `enable_data`, stacked pause+backpressure disables, control channels staying unblocked mid-pause, pre-registered and normally-registered baselines, and two threaded stress tests that race first-time registrations against disable/enable toggles and assert exact element counts. The whole file passes with 25 passed and 1 xfailed. The full existing Python unit-test tree under `amber/src/test/python/core` was also run before and after the change, and the pass/fail sets are identical apart from the new tests, so no existing behavior changed. ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) -- 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]
