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

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)

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)

// 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

// 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.

3.2 KiB Raw Blame History Unescape Escape