Musxeto opened a new pull request, #7850:
URL: https://github.com/apache/texera/pull/7850
### Problem
`LinkedBlockingMultiQueue.add_sub_queue` diverged from the upstream Java
`LinkedBlockingMultiQueue 0.6.0` it ports. The Java implementation uses
`subQueues.putIfAbsent(key, subQueue)`, which leaves the map untouched when
the key already exists. The Python port unconditionally overwrote
`self.sub_queues[key]` with a freshly constructed `SubQueue` before checking
whether the key was already present.
On a repeated key, the method still returned the old `SubQueue` (matching its
docstring), but a new, unattached `SubQueue` (`priority_group = None`) had
already been installed in the map. The failure chain was:
Before: `add_sub_queue("k")` twice → orphan `SubQueue` in map
(priority_group = None)
After: `put("k", item)` → `total_count` incremented, `get()` woken up
→ `get_next()` walks `priority_groups`, finds no non-empty group,
returns `None`
→ `None.dequeue()` → `AttributeError` while holding `take_lock`
The element was also permanently lost to consumers.
### Changes
**`amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py`**
Early-return from `add_sub_queue` when the key is already present, leaving
the map and all priority groups untouched. The `SubQueue` object is now
constructed only when the key is absent, matching the upstream Java
semantics.
**`amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py`**
Added `TestAddSubQueue` with 5 tests (written red-first per TDD):
- `test_new_key_returns_none` — positive path: first registration returns
`None`
- `test_repeated_key_returns_existing_sub_queue` — duplicate returns the
original object
- `test_repeated_key_keeps_existing_queue_in_map` — map holds the original
after a second add
- `test_repeated_key_priority_group_is_not_none` — `priority_group` is not
`None` after duplicate add
- `test_repeated_key_put_then_get_does_not_crash` — exact reproduction from
the issue
### How was this PR tested?
```
/venv312/bin/pytestsrc/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py
-v
```
Result: **47 passed** (5 new + 42 pre-existing, no regressions).
### Was this PR authored or co-authored using generative AI tooling?
no
Closes #7810
--
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]
