`LinkedBlockingMultiQueue.add_sub_queue` replaces an existing sub-queue instead of keeping it (putIfAbsent divergence)

### 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'
```