mengw15 opened a new issue, #7810: URL: https://github.com/apache/texera/issues/7810
### What happened? `LinkedBlockingMultiQueue.add_sub_queue` (`amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py:412-414`) diverges from the upstream Java `LinkedBlockingMultiQueue` 0.6.0 it ports, which registers with `subQueues.putIfAbsent(key, subQueue)`. The Python port unconditionally overwrites `self.sub_queues[key]` with a freshly created `SubQueue` before checking whether the key already exists. On a repeated key the method still returns the previous `SubQueue` (matching its docstring), but it has already installed a replacement that belongs to no priority group (`priority_group is None`). The next `put` on that key stores the element into the unattached sub-queue and increments `total_count`, which wakes a blocked `get()`; `DefaultSubQueueSelection.get_next()` walks `priority_groups`, finds no non-empty group, returns `None`, and `get()` raises `AttributeError: 'NoneType' object has no attribute 'dequeue'` while holding `take_lock`. The element is also permanently lost to consumers. Expected: a repeated `add_sub_queue` keeps the existing registered sub-queue and returns it, as upstream does. Not reachable through the worker today: `InternalQueue` deduplicates registrations via `_queue_ids` (`amber/src/main/python/core/models/internal_queue.py:79`), so this only bites direct users of the class. Found during review of #6906 (https://github.com/apache/texera/pull/6906#discussion_r3738766123) — same family of port divergences as #6903. ### How to reproduce? ```python from core.util.customized_queue.linked_blocking_multi_queue import ( LinkedBlockingMultiQueue, ) q = LinkedBlockingMultiQueue() q.add_sub_queue("k", 0) q.add_sub_queue("k", 0) # repeated key: overwrites the map with an unattached SubQueue q.put("k", "x") q.get() # AttributeError: 'NoneType' object has no attribute 'dequeue' ``` Verified at #6906's head (41b866adc): the second `add_sub_queue` returns the old sub-queue but `q.get_sub_queue("k").priority_group` is `None`, and `get()` raises. ### Version/Branch 1.3.0-incubating-SNAPSHOT (main) ### Commit Hash (Optional) 41b866adc (also present on current main) ### Relevant log output ```shell AttributeError: 'NoneType' object has no attribute 'dequeue' ``` -- 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]
