# UNDF: UNDF-2026-000000507
# pulsar-0003 — PersistentTopic: replicationClusters List.contains() O(C×R) in replication-check hot path

## Metadata
- **Project**: Apache Pulsar
- **Component**: `pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java`
- **CWE**: CWE-407 (Inefficient Algorithmic Complexity)
- **Severity**: MEDIUM
- **Complexity**: O(C × R) → O(R) where C = configured clusters, R = active replicators

## Location

```
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
```

## Defective code

### Site 1 — removeOrphanReplicationCursors() line 549–557

```java
List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
for (ManagedCursor cursor : ledger.getCursors()) {
    if (cursor.getName().startsWith(replicatorPrefix)) {
        String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
        if (!replicationClusters.contains(remoteCluster)) {  // O(C) List.contains per cursor
            futures.add(removeReplicator(remoteCluster));
        }
    }
}
```

### Site 2 — checkReplication() line 1990–1997 (hot path, called periodically per topic)

```java
List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
// ...
replicators.forEach((cluster, replicator) -> {
    ((PersistentReplicator) replicator).updateMessageTTL(newMessageTTLInSeconds);
    if (!cluster.equals(localCluster)) {
        if (!configuredClusters.contains(cluster)) {  // O(C) List.contains per replicator
            futures.add(removeReplicator(cluster));
        }
    }
});
```

`topicPolicies.getReplicationClusters().get()` returns `List<String>` (via `PolicyHierarchyValue<List<String>>`).
Total cost: O(R × C) where R = number of active replicators, C = number of configured clusters.

### Contrast with NonPersistentTopic (already fixed correctly)

```java
// NonPersistentTopic.java line 595 — correct approach:
Set<String> configuredClusters = new HashSet<>(topicPolicies.getReplicationClusters().get());
```

`NonPersistentTopic` wraps in `HashSet<>` before the loop; `PersistentTopic` does not.

## Fix

```java
// Site 1 — removeOrphanReplicationCursors():
// BEFORE:
List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
// AFTER:
Set<String> replicationClusters = new HashSet<>(topicPolicies.getReplicationClusters().get());

// Site 2 — checkReplication():
// BEFORE:
List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
// AFTER:
Set<String> configuredClusters = new HashSet<>(topicPolicies.getReplicationClusters().get());
```

One-line fix at each site; matches the pattern already used in `NonPersistentTopic`.

## Complexity analysis

| Scenario | Before | After |
|----------|--------|-------|
| R replicators, C clusters | O(R × C) | O(R + C) |
| R=20, C=20 | 400 ops per topic check | 40 ops |
| Geo-replicated namespace with many topics | Compounds per-topic × per-replication-check interval | Minimal |

## Notes

`checkReplication()` is invoked periodically for every topic by the broker. In a large geo-replicated deployment with many topics and many clusters, the O(R × C) cost per check compounds across all topics. The fix mirrors the approach already used in `NonPersistentTopic`.
