88 lines
3.3 KiB
Markdown
88 lines
3.3 KiB
Markdown
# 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`.
|