# UNDF: UNDF-2026-000000509
# pulsar-0005 — PartialRoundRobinMessageRouterImpl CopyOnWriteArrayList.contains O(N×L) → HashSet O(N)

## Classification

| Field       | Value |
|-------------|-------|
| CWE         | CWE-407 Inefficient Algorithmic Complexity |
| Severity    | MEDIUM |
| Component   | `pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java` |
| Introduced  | PartialRoundRobin routing support |
| Status      | PATCHED (unit test PASS) |

## Defect

In `getOrCreatePartialList()`, when the partial list needs to grow, a stream over all N partition
indices filters out those already present in `partialList`, a `CopyOnWriteArrayList<Integer>`:

```java
// DEFECTIVE — PartialRoundRobinMessageRouterImpl.java:73-74
} else if (partialList.size() < numPartitionsLimit && partialList.size() < metadata.numPartitions()) {
    partialList.addAll(IntStream.range(0, metadata.numPartitions()).boxed()
            .filter(e -> !partialList.contains(e))   // O(L) per element
            .collect(Collectors.collectingAndThen(Collectors.toList(), list -> {
```

`CopyOnWriteArrayList.contains()` is O(L) where L = `partialList.size()` (up to `numPartitionsLimit`).
The stream visits N elements (N = `metadata.numPartitions()`).
Total: **O(N × L)**.

For a topic with N=1 000 partitions and L=500 already selected, this performs 500 000 comparisons
instead of 1 000.

Additionally, `partialList` is a `CopyOnWriteArrayList`, which copies the backing array on every
`add()` during the `addAll()` batch, adding O(L²) write overhead if `addAll()` is not atomic.
In practice `addAll()` is a single copy, but `contains()` remains O(L).

## Fix

Build a `HashSet<Integer>` snapshot of `partialList` before the filter, then use the set for
O(1) membership tests:

```java
// FIXED
Set<Integer> existing = new HashSet<>(partialList);
partialList.addAll(IntStream.range(0, metadata.numPartitions()).boxed()
        .filter(e -> !existing.contains(e))   // O(1) per element
        .collect(Collectors.collectingAndThen(Collectors.toList(), list -> {
            Collections.shuffle(list);
            return list.stream();
        })).limit(numPartitionsLimit - partialList.size()).collect(Collectors.toList()));
```

The snapshot is built once in O(L) and all N filter calls are O(1), giving O(N + L) total.

## Complexity

| Scenario | Before | After | Ratio |
|----------|--------|-------|-------|
| N=100, L=50 | O(5 000) | O(150) | 33× |
| N=500, L=250 | O(125 000) | O(750) | 167× |
| N=1 000, L=500 | O(500 000) | O(1 500) | 333× |

## Speedup (measured)

See unit test `PulsarPartialRoundRobinRouterTest.java`.
Measured ratio ≥ 5× at N=500 (SLOW CopyOnWriteArrayList vs FAST HashSet snapshot).

## Affected File

```
pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java
  Line 35:  partialList = new CopyOnWriteArrayList<>()  — fine for reads, O(L) contains
  Line 74:  .filter(e -> !partialList.contains(e))      — O(N × L), should snapshot to HashSet first
```
