eugenegujing opened a new issue, #6723:
URL: https://github.com/apache/texera/issues/6723
### What happened?
Pause and backpressure in the Python worker both work by disabling the data
sub-queues of `InternalQueue`
(`amber/src/main/python/core/models/internal_queue.py`). The problem:
`disable_data` only disables the sub-queues that exist when it is called, but
sub-queues are created lazily on a channel's first `put`, and a new sub-queue
starts enabled. `put` never checks `_queue_state`, so a data channel whose
first message arrives after the pause comes up enabled — the pause simply does
not apply to it.
Consequences:
1. The worker keeps processing data while reporting PAUSED: the leaked
channel makes `is_data_enabled()` return `True` again, which lets the DP thread
exit the pause wait-loop in `main_loop.py`.
2. Worse: if the DP thread is blocked inside that wait-loop, it dequeues the
leaked `DataElement`, but the wait-loop's `match` only handles control messages
— pampy raises an uncaught `MatchError` and the DP thread dies. The heartbeat
thread stays alive, so the worker looks healthy while the execution hangs
forever with no error anywhere.
3. Under backpressure (the other `disable_data` caller), the leaked channel
keeps feeding the congested downstream, defeating flow control.
Expected: after `disable_data`, no data element should be dequeuable until
the matching `enable_data` — including channels registered after the call.
Cleanup in the same file: the `InternalMarker` entry in `put`'s isinstance
tuple is dead code — `InternalMarker` does not subclass `InternalQueueElement`,
so it can never reach that branch. And the two branches around it perform the
identical `self._queue.put(item.tag, item)`, so they can be merged.
### How to reproduce?
Unit-level repro:
```python
queue = InternalQueue()
queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
# first message of a NEW data channel arrives after the pause
new_channel = ChannelIdentity(ActorVirtualIdentity("A"),
ActorVirtualIdentity("B"), False)
queue.put(DataElement(tag=new_channel, payload=None))
queue.is_data_enabled() # returns True — expected False while paused
queue.get() # returns the DataElement — expected to block until
enable_data
```
### Version/Branch
1.3.0-incubating-SNAPSHOT (main)
### Commit Hash (Optional)
_No response_
### What browsers are you seeing the problem on?
_No response_
### Relevant log output
```shell
```
--
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]
