pulsar: 3 CWE-407 defects — load-manager/replicator/namespace List.contains O(N^2)
pulsar-0001: ModularLoadManagerImpl.reapDeadBrokerPreallocations() receives aliveBrokers as List<String> from listLocks(); .contains() inside O(B) loop → O(B^2); fix: HashSet wrap before loop. pulsar-0002: PersistentTopic.removeOrphanReplicationCursors() and checkReplicationStatus() call configuredClusters.contains() (List<String>) inside loops over cursors and replicators → O(C×R); fix: HashSet wrap. pulsar-0003: NamespacesBase.internalGetTopicHashPositionsAsync() calls allTopicsInThisBundle.contains() (List<String>) inside for loop over query topics → O(T×B); fix: HashSet wrap before loop. UNDF-2026-000000505 through UNDF-2026-000000507; 3/3 unit tests PASS.
This commit is contained in:
parent
e72461aba9
commit
b0a9a83efc
6 changed files with 351 additions and 1 deletions
|
|
@ -737,5 +737,7 @@
|
|||
"spring-framework-0003": "UNDF-2026-000000736",
|
||||
"dragonfly-0002": "UNDF-2026-000000737",
|
||||
"kafka-0002": "UNDF-2026-000000738",
|
||||
"kafka-0003": "UNDF-2026-000000739"
|
||||
"kafka-0003": "UNDF-2026-000000739",
|
||||
"micronaut-core-0001": "UNDF-2026-000000740",
|
||||
"micronaut-core-0002": "UNDF-2026-000000741"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,16 @@
|
|||
# UNDF: UNDF-2026-000000505
|
||||
# UNDF:
|
||||
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
|
||||
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
|
||||
@@ -320,7 +320,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
|
||||
|
||||
// For each broker that we have a recent load report, see if they are still alive
|
||||
- private void reapDeadBrokerPreallocations(List<String> aliveBrokers) {
|
||||
+ private void reapDeadBrokerPreallocations(List<String> aliveBrokers) {
|
||||
+ Set<String> aliveBrokerSet = new HashSet<>(aliveBrokers);
|
||||
for (String broker : loadData.getBrokerData().keySet()) {
|
||||
- if (!aliveBrokers.contains(broker)) {
|
||||
+ if (!aliveBrokerSet.contains(broker)) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Broker {} appears to have stopped; now reclaiming any preallocations", broker);
|
||||
}
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
# UNDF: UNDF-2026-000000506
|
||||
# UNDF:
|
||||
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
|
||||
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
|
||||
@@ -547,9 +547,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
|
||||
private CompletableFuture<Void> removeOrphanReplicationCursors() {
|
||||
List<CompletableFuture<Void>> futures = new ArrayList<>();
|
||||
- List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
|
||||
+ Set<String> replicationClusters = new HashSet<>(topicPolicies.getReplicationClusters().get());
|
||||
for (ManagedCursor cursor : ledger.getCursors()) {
|
||||
if (cursor.getName().startsWith(replicatorPrefix)) {
|
||||
String remoteCluster = PersistentReplicator.getRemoteCluster(cursor.getName());
|
||||
if (!replicationClusters.contains(remoteCluster)) {
|
||||
log.warn("Remove the orphan replicator because the cluster '{}' does not exist", remoteCluster);
|
||||
futures.add(removeReplicator(remoteCluster));
|
||||
}
|
||||
}
|
||||
}
|
||||
return FutureUtil.waitForAll(futures);
|
||||
}
|
||||
|
||||
@@ -1954,7 +1954,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
|
||||
- List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
|
||||
+ Set<String> configuredClusters = new HashSet<>(topicPolicies.getReplicationClusters().get());
|
||||
if (CollectionUtils.isEmpty(configuredClusters)) {
|
||||
log.warn("[{}] No replication clusters configured", name);
|
||||
return CompletableFuture.completedFuture(null);
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
# UNDF: UNDF-2026-000000507
|
||||
# UNDF:
|
||||
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
|
||||
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
|
||||
@@ -1530,7 +1530,8 @@ public abstract class NamespacesBase extends PulsarWebResource {
|
||||
.thenApply(allTopicsInThisBundle -> {
|
||||
Map<String, Long> topicHashPositions = new HashMap<>();
|
||||
+ Set<String> allTopicsSet = new HashSet<>(allTopicsInThisBundle);
|
||||
if (topics == null || topics.size() == 0) {
|
||||
allTopicsInThisBundle.forEach(t -> {
|
||||
topicHashPositions.put(t,
|
||||
pulsar().getNamespaceService().getNamespaceBundleFactory()
|
||||
.getLongHashCode(t));
|
||||
});
|
||||
} else {
|
||||
for (String topic : topics.stream().map(Codec::decode).toList()) {
|
||||
TopicName topicName = TopicName.get(topic);
|
||||
// partitioned topic
|
||||
if (topicName.getPartitionIndex() == -1) {
|
||||
allTopicsInThisBundle.stream()
|
||||
.filter(t -> TopicName.get(t).getPartitionedTopicName()
|
||||
.equals(TopicName.get(topic).getPartitionedTopicName()))
|
||||
.forEach(partition -> {
|
||||
topicHashPositions.put(partition,
|
||||
pulsar().getNamespaceService()
|
||||
.getNamespaceBundleFactory()
|
||||
.getLongHashCode(partition));
|
||||
});
|
||||
} else { // topic partition
|
||||
- if (allTopicsInThisBundle.contains(topicName.toString())) {
|
||||
+ if (allTopicsSet.contains(topicName.toString())) {
|
||||
topicHashPositions.put(topic,
|
||||
pulsar().getNamespaceService().getNamespaceBundleFactory()
|
||||
.getLongHashCode(topic));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
BIN
defects/pulsar/unit/PulsarTest.class
Normal file
BIN
defects/pulsar/unit/PulsarTest.class
Normal file
Binary file not shown.
267
defects/pulsar/unit/PulsarTest.java
Normal file
267
defects/pulsar/unit/PulsarTest.java
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
import java.util.*;
|
||||
|
||||
/**
|
||||
* CWE-407 unit tests for Apache Pulsar.
|
||||
*
|
||||
* Three defects measured:
|
||||
* pulsar-0001: ModularLoadManagerImpl.reapDeadBrokerPreallocations()
|
||||
* aliveBrokers is List<String> from listLocks(); O(B^2) lookup
|
||||
* pulsar-0002: PersistentTopic.removeOrphanReplicationCursors() and
|
||||
* checkReplicationStatus(); configuredClusters is List<String>
|
||||
* and .contains() is called inside a loop over cursors/replicators
|
||||
* pulsar-0003: NamespacesBase.internalGetTopicHashPositionsAsync();
|
||||
* allTopicsInThisBundle is List<String> from getOwnedTopicListForNamespaceBundle();
|
||||
* .contains() called inside a for loop over topics query argument
|
||||
*/
|
||||
public class PulsarTest {
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// pulsar-0001: reapDeadBrokerPreallocations O(B^2)
|
||||
//
|
||||
// Pattern:
|
||||
// for (String broker : loadData.getBrokerData().keySet()) {
|
||||
// if (!aliveBrokers.contains(broker)) { ... } // aliveBrokers is List
|
||||
// }
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
static long defectReapDeadBrokers(List<String> knownBrokers, List<String> aliveBrokers) {
|
||||
// Simulates: for each known broker, scan aliveBrokers list for membership
|
||||
long ops = 0;
|
||||
List<String> dead = new ArrayList<>();
|
||||
for (String broker : knownBrokers) {
|
||||
ops++;
|
||||
if (!aliveBrokers.contains(broker)) { // O(N) linear scan per broker
|
||||
dead.add(broker);
|
||||
}
|
||||
}
|
||||
return ops;
|
||||
}
|
||||
|
||||
static long fixedReapDeadBrokers(List<String> knownBrokers, List<String> aliveBrokers) {
|
||||
Set<String> aliveSet = new HashSet<>(aliveBrokers);
|
||||
long ops = 0;
|
||||
List<String> dead = new ArrayList<>();
|
||||
for (String broker : knownBrokers) {
|
||||
ops++;
|
||||
if (!aliveSet.contains(broker)) { // O(1) hash lookup
|
||||
dead.add(broker);
|
||||
}
|
||||
}
|
||||
return ops;
|
||||
}
|
||||
|
||||
static void testPulsar0001() {
|
||||
int B = 500; // number of known brokers in loadData
|
||||
List<String> knownBrokers = new ArrayList<>();
|
||||
List<String> aliveBrokers = new ArrayList<>();
|
||||
for (int i = 0; i < B; i++) {
|
||||
knownBrokers.add("broker-" + i);
|
||||
}
|
||||
// Half are alive (worst-case scan length for the rest)
|
||||
for (int i = 0; i < B / 2; i++) {
|
||||
aliveBrokers.add("broker-" + i);
|
||||
}
|
||||
|
||||
int REPS = 200;
|
||||
|
||||
// Warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defectReapDeadBrokers(knownBrokers, aliveBrokers);
|
||||
fixedReapDeadBrokers(knownBrokers, aliveBrokers);
|
||||
}
|
||||
|
||||
long t0 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
defectReapDeadBrokers(knownBrokers, aliveBrokers);
|
||||
}
|
||||
long defectNs = System.nanoTime() - t0;
|
||||
|
||||
long t1 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
fixedReapDeadBrokers(knownBrokers, aliveBrokers);
|
||||
}
|
||||
long fixedNs = System.nanoTime() - t1;
|
||||
|
||||
double ratio = (double) defectNs / fixedNs;
|
||||
System.out.printf("pulsar-0001 reapDeadBrokerPreallocations B=%d defect=%.1fms fixed=%.1fms ratio=%.1fx%n",
|
||||
B,
|
||||
defectNs / 1e6 / REPS,
|
||||
fixedNs / 1e6 / REPS,
|
||||
ratio);
|
||||
|
||||
if (ratio < 3.0) {
|
||||
throw new RuntimeException("pulsar-0001 FAIL: expected ratio >= 3.0, got " + ratio);
|
||||
}
|
||||
System.out.println("pulsar-0001 PASS");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// pulsar-0002: PersistentTopic replication cluster list scan O(C*R)
|
||||
//
|
||||
// Pattern (removeOrphanReplicationCursors):
|
||||
// List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
|
||||
// for (ManagedCursor cursor : ledger.getCursors()) {
|
||||
// if (!replicationClusters.contains(remoteCluster)) { ... } // O(C)
|
||||
// }
|
||||
//
|
||||
// Pattern (checkReplicationStatus):
|
||||
// List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
|
||||
// replicators.forEach((cluster, replicator) -> {
|
||||
// if (!configuredClusters.contains(cluster)) { ... } // O(C)
|
||||
// });
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
static long defectReplicationCursorScan(List<String> replicationClusters, List<String> cursors) {
|
||||
long removes = 0;
|
||||
for (String cursor : cursors) {
|
||||
// simulates: if (!replicationClusters.contains(remoteCluster))
|
||||
if (!replicationClusters.contains(cursor)) {
|
||||
removes++;
|
||||
}
|
||||
}
|
||||
return removes;
|
||||
}
|
||||
|
||||
static long fixedReplicationCursorScan(List<String> replicationClusters, List<String> cursors) {
|
||||
Set<String> clusterSet = new HashSet<>(replicationClusters);
|
||||
long removes = 0;
|
||||
for (String cursor : cursors) {
|
||||
if (!clusterSet.contains(cursor)) {
|
||||
removes++;
|
||||
}
|
||||
}
|
||||
return removes;
|
||||
}
|
||||
|
||||
static void testPulsar0002() {
|
||||
int C = 500; // configured replication clusters (list size)
|
||||
int R = 500; // active replicator cursors (loop iterations)
|
||||
// Simulate geo-replicated topic: R cursors, C configured clusters
|
||||
List<String> replicationClusters = new ArrayList<>();
|
||||
List<String> cursorClusters = new ArrayList<>();
|
||||
for (int i = 0; i < C; i++) {
|
||||
replicationClusters.add("cluster-" + i);
|
||||
}
|
||||
// cursors: all are orphans — each triggers full linear scan before miss
|
||||
for (int i = 0; i < R; i++) {
|
||||
cursorClusters.add("orphan-cluster-" + i); // not in list: O(C) scan
|
||||
}
|
||||
|
||||
int REPS = 500;
|
||||
|
||||
// Warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defectReplicationCursorScan(replicationClusters, cursorClusters);
|
||||
fixedReplicationCursorScan(replicationClusters, cursorClusters);
|
||||
}
|
||||
|
||||
long t0 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
defectReplicationCursorScan(replicationClusters, cursorClusters);
|
||||
}
|
||||
long defectNs = System.nanoTime() - t0;
|
||||
|
||||
long t1 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
fixedReplicationCursorScan(replicationClusters, cursorClusters);
|
||||
}
|
||||
long fixedNs = System.nanoTime() - t1;
|
||||
|
||||
double ratio = (double) defectNs / fixedNs;
|
||||
System.out.printf("pulsar-0002 replicationCluster list scan C=%d R=%d defect=%.2fms fixed=%.2fms ratio=%.1fx%n",
|
||||
C, R,
|
||||
defectNs / 1e6 / REPS,
|
||||
fixedNs / 1e6 / REPS,
|
||||
ratio);
|
||||
|
||||
if (ratio < 3.0) {
|
||||
throw new RuntimeException("pulsar-0002 FAIL: expected ratio >= 3.0, got " + ratio);
|
||||
}
|
||||
System.out.println("pulsar-0002 PASS");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// pulsar-0003: NamespacesBase getTopicHashPositions O(T*B)
|
||||
//
|
||||
// Pattern:
|
||||
// List<String> allTopicsInThisBundle = getOwnedTopicListForNamespaceBundle(nsBundle);
|
||||
// for (String topic : topics) {
|
||||
// if (allTopicsInThisBundle.contains(topicName.toString())) { ... } // O(B)
|
||||
// }
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
static long defectTopicHashPositions(List<String> allTopicsInBundle, List<String> queryTopics) {
|
||||
long hits = 0;
|
||||
for (String topic : queryTopics) {
|
||||
// simulates: if (allTopicsInThisBundle.contains(topicName.toString()))
|
||||
if (allTopicsInBundle.contains(topic)) {
|
||||
hits++;
|
||||
}
|
||||
}
|
||||
return hits;
|
||||
}
|
||||
|
||||
static long fixedTopicHashPositions(List<String> allTopicsInBundle, List<String> queryTopics) {
|
||||
Set<String> topicSet = new HashSet<>(allTopicsInBundle);
|
||||
long hits = 0;
|
||||
for (String topic : queryTopics) {
|
||||
if (topicSet.contains(topic)) {
|
||||
hits++;
|
||||
}
|
||||
}
|
||||
return hits;
|
||||
}
|
||||
|
||||
static void testPulsar0003() {
|
||||
int B = 1000; // topics in bundle
|
||||
int T = 500; // topics in query list
|
||||
List<String> allTopicsInBundle = new ArrayList<>();
|
||||
List<String> queryTopics = new ArrayList<>();
|
||||
for (int i = 0; i < B; i++) {
|
||||
allTopicsInBundle.add("persistent://tenant/ns/topic-" + i);
|
||||
}
|
||||
// Query includes topics that land near the end of the bundle list (worst-case scan)
|
||||
for (int i = 0; i < T; i++) {
|
||||
queryTopics.add("persistent://tenant/ns/topic-" + (B - T + i));
|
||||
}
|
||||
|
||||
int REPS = 200;
|
||||
|
||||
// Warm up
|
||||
for (int i = 0; i < 5; i++) {
|
||||
defectTopicHashPositions(allTopicsInBundle, queryTopics);
|
||||
fixedTopicHashPositions(allTopicsInBundle, queryTopics);
|
||||
}
|
||||
|
||||
long t0 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
defectTopicHashPositions(allTopicsInBundle, queryTopics);
|
||||
}
|
||||
long defectNs = System.nanoTime() - t0;
|
||||
|
||||
long t1 = System.nanoTime();
|
||||
for (int r = 0; r < REPS; r++) {
|
||||
fixedTopicHashPositions(allTopicsInBundle, queryTopics);
|
||||
}
|
||||
long fixedNs = System.nanoTime() - t1;
|
||||
|
||||
double ratio = (double) defectNs / fixedNs;
|
||||
System.out.printf("pulsar-0003 topicHashPositions list scan B=%d T=%d defect=%.1fms fixed=%.1fms ratio=%.1fx%n",
|
||||
B, T,
|
||||
defectNs / 1e6 / REPS,
|
||||
fixedNs / 1e6 / REPS,
|
||||
ratio);
|
||||
|
||||
if (ratio < 5.0) {
|
||||
throw new RuntimeException("pulsar-0003 FAIL: expected ratio >= 5.0, got " + ratio);
|
||||
}
|
||||
System.out.println("pulsar-0003 PASS");
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
testPulsar0001();
|
||||
testPulsar0002();
|
||||
testPulsar0003();
|
||||
System.out.println("All Pulsar CWE-407 tests PASS");
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue