# UNDF: UNDF-2026-000000611
# pulsar-0006: PartialRoundRobinMessageRouterImpl.getOrCreatePartialList — O(P²) CopyOnWriteArrayList.contains during partition expansion

## CWE-407 — Algorithmic Complexity

| Field | Value |
|-------|-------|
| ID | pulsar-0006 |
| Severity | MEDIUM |
| Ecosystem | Apache Pulsar |
| Package | `org.apache.pulsar.client.impl.customroute` |
| File | `pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java` |
| Lines | 72–78 |
| Complexity | O(P²) — CopyOnWriteArrayList.contains called per element of IntStream.range(0, numPartitions) |
| Hot path | Triggered every time a Pulsar topic adds partitions while a producer using this router is active (partition expansion event) |

## Defect

```java
// partialList is CopyOnWriteArrayList<Integer>
private final List<Integer> partialList = new CopyOnWriteArrayList<>();

private List<Integer> getOrCreatePartialList(TopicMetadata metadata) {
    if (partialList.isEmpty()
            || partialList.size() < numPartitionsLimit
               && partialList.size() < metadata.numPartitions()) {
        synchronized (this) {
            // ...
            } else if (partialList.size() < numPartitionsLimit
                    && partialList.size() < metadata.numPartitions()) {
                // DEFECT: stream over all P partition IDs, calling partialList.contains(e)
                // for each — O(P) per call × O(P) elements = O(P²)
                partialList.addAll(IntStream.range(0, metadata.numPartitions()).boxed()
                        .filter(e -> !partialList.contains(e))  // O(P) per element
                        .collect(Collectors.collectingAndThen(Collectors.toList(), list -> {
                            Collections.shuffle(list);
                            return list.stream();
                        })).limit(numPartitionsLimit - partialList.size())
                        .collect(Collectors.toList()));
            }
        }
    }
    return partialList;
}
```

`P` = `metadata.numPartitions()`. Pulsar topics with thousands of partitions are common in high-throughput deployments. CopyOnWriteArrayList.contains is O(N) — there is no hash-based shortcut.

## Fix

```java
// AFTER: snapshot partialList to a HashSet once, then filter in O(1) per element
} else if (partialList.size() < numPartitionsLimit
        && partialList.size() < metadata.numPartitions()) {
    Set<Integer> existing = new HashSet<>(partialList);  // O(P) one-time build
    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()));
}
```

## Speedup

| P (total partitions) | Before (ops) | After (ops) | Speedup |
|----------------------|--------------|-------------|---------|
| 100 | 10,000 | 100 | 100× |
| 1,000 | 1,000,000 | 1,000 | 1,000× |
| 4,000 | 16,000,000 | 4,000 | 4,000× |

## Notes

The synchronized block prevents concurrent re-entrancy, but the inner O(P²) work still stalls all threads waiting on the lock during a partition expansion event. In a high-throughput producer this synchronized stall causes a latency spike proportional to P².
