From b0a9a83efc791a70aedd6787ed31753e937cef35 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 30 Mar 2026 08:43:51 -0400 Subject: [PATCH] =?UTF-8?q?pulsar:=203=20CWE-407=20defects=20=E2=80=94=20l?= =?UTF-8?q?oad-manager/replicator/namespace=20List.contains=20O(N^2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pulsar-0001: ModularLoadManagerImpl.reapDeadBrokerPreallocations() receives aliveBrokers as List 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) inside loops over cursors and replicators → O(C×R); fix: HashSet wrap. pulsar-0003: NamespacesBase.internalGetTopicHashPositionsAsync() calls allTopicsInThisBundle.contains() (List) 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. --- UNDF-REGISTRY.json | 4 +- ...nager-reap-dead-broker-list-contains.patch | 16 ++ ...ic-replication-cluster-list-contains.patch | 27 ++ ...e-topic-hash-positions-list-contains.patch | 38 +++ defects/pulsar/unit/PulsarTest.class | Bin 0 -> 5221 bytes defects/pulsar/unit/PulsarTest.java | 267 ++++++++++++++++++ 6 files changed, 351 insertions(+), 1 deletion(-) create mode 100644 defects/pulsar/patch/pulsar-0001-modular-load-manager-reap-dead-broker-list-contains.patch create mode 100644 defects/pulsar/patch/pulsar-0002-persistent-topic-replication-cluster-list-contains.patch create mode 100644 defects/pulsar/patch/pulsar-0003-namespaces-base-topic-hash-positions-list-contains.patch create mode 100644 defects/pulsar/unit/PulsarTest.class create mode 100644 defects/pulsar/unit/PulsarTest.java diff --git a/UNDF-REGISTRY.json b/UNDF-REGISTRY.json index e4afb0f55..7ecf75915 100644 --- a/UNDF-REGISTRY.json +++ b/UNDF-REGISTRY.json @@ -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" } diff --git a/defects/pulsar/patch/pulsar-0001-modular-load-manager-reap-dead-broker-list-contains.patch b/defects/pulsar/patch/pulsar-0001-modular-load-manager-reap-dead-broker-list-contains.patch new file mode 100644 index 000000000..93aa1d0ca --- /dev/null +++ b/defects/pulsar/patch/pulsar-0001-modular-load-manager-reap-dead-broker-list-contains.patch @@ -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 aliveBrokers) { ++ private void reapDeadBrokerPreallocations(List aliveBrokers) { ++ Set 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); + } diff --git a/defects/pulsar/patch/pulsar-0002-persistent-topic-replication-cluster-list-contains.patch b/defects/pulsar/patch/pulsar-0002-persistent-topic-replication-cluster-list-contains.patch new file mode 100644 index 000000000..d5b1603f4 --- /dev/null +++ b/defects/pulsar/patch/pulsar-0002-persistent-topic-replication-cluster-list-contains.patch @@ -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 removeOrphanReplicationCursors() { + List> futures = new ArrayList<>(); +- List replicationClusters = topicPolicies.getReplicationClusters().get(); ++ Set 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 configuredClusters = topicPolicies.getReplicationClusters().get(); ++ Set configuredClusters = new HashSet<>(topicPolicies.getReplicationClusters().get()); + if (CollectionUtils.isEmpty(configuredClusters)) { + log.warn("[{}] No replication clusters configured", name); + return CompletableFuture.completedFuture(null); diff --git a/defects/pulsar/patch/pulsar-0003-namespaces-base-topic-hash-positions-list-contains.patch b/defects/pulsar/patch/pulsar-0003-namespaces-base-topic-hash-positions-list-contains.patch new file mode 100644 index 000000000..06c188f5c --- /dev/null +++ b/defects/pulsar/patch/pulsar-0003-namespaces-base-topic-hash-positions-list-contains.patch @@ -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 topicHashPositions = new HashMap<>(); ++ Set 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)); + } + } + } + } diff --git a/defects/pulsar/unit/PulsarTest.class b/defects/pulsar/unit/PulsarTest.class new file mode 100644 index 0000000000000000000000000000000000000000..f44e2be639c845a5677bc97e1fec43ec50a49a61 GIT binary patch literal 5221 zcmb`Le{>vG7017Cf6e}wG|e`fHVG*cXrXC-HiZO|Hj<1Jo$ zoh>aWzZygf6huV|2v~ul{N^A{i$C$`LC--&{X+%ML600jg@eHHNTIFYH#13gn-nQbZt_w2Z87 znOT8~1$F(C(|zvy&Irm7RWM6MIbs6QsUiZ(Rz0`M*g<`?XoFlaTg4ny3i#6Y>+*5k z1%7lYs4j9ZZe^3{Ez;r|6?1Vits6GemYz)K1Y!#u;|dkXPeo9x;uI`B`tX?yfVn~x_y0dO=1a+uUuu!06QUg~4#>1oS0u+NR3ZKC5$XBaJ5q@Y#BVw@om9>}M1dUnu|v5zH; z&HVTpL(eQT^h8hA+-78{>C{3CbDGw(bbdn999NglZ(6i@gfDH2C!UyqNz;DXF+Ux~ zxmd2?JQe4oZ+tWDkoKi5V~fE|RkrJ?ys>(-%-N!#yAqv&?xxXexN8VwB~~d|EtL%v zMPZql-;|>3u?5Qtx6OIYZCiu5f zQYuEIT$P)W(})FO3T&J-p_TbFDVEAJIkQ@dS+C`&RMWIxnLulbv#7Nwi=3yTEQ-R1 zi-TaY6nNXJNpGVJknL1n#dc}z9LK3aGm{*a1!KU>CGF&u9_>)EQwq(tJ(?;L#Fa(v zEY_?kdWzh!c^E8q=VPG=oKx;3NfZI;*EL0cEw=rV)?TM#mpn9PSgbHkt&xbzsq0mI zVd_+?ocf}QFHN0V9K?+RHQlL{=4jG-*PYvVMoYUUmFMhlvcUJ631U>*pG+I8@*|s! z?4WLgRT8)K;cYARjQbplC%2?^E1#wG1;-efStxVktnlQp6Le_#p<|a~} zMK^Bdv%|*nq~yKS@kLhPCdnED*<+ek&dTbUm4>y|Oyq+2I;-O*yBs%)AodD0P3G$& zZFzTJe}`u5$gp@D3C+$st!t^axT&Q<+hST8*W4u#8g`Ods`v&!V`ej3^>kyw+k1-? zw|*kU-xQde;W0r#jI`C!+{}RoP;)xhY$u-l+P8}IEjhH56Keq zZGm}{l_;6s&ZDW>38{1VE!T(HH0%_(qp;4nf=(RI*VBoVk(<|Vn%nXjcQ*^q#)3UB zS~$2fV@xAnxVug#)LB@$T@}4NeBx|sItA3eblS-Fru1CS$SHV4VEz=JrZZK+V+`^~ zi?aGVexa5Cd2Bc%EYNJ(c)f|wH5>&;|H1_xr2s}T4RsyYH%SnsZ{3>?Hs9^8qNN}s&fJXp8? z*o)euROPWhb_zcuE%C6Oir{4SHySs$7H6?z>cKpowDYlt5;NgFidjAd5d~#Uhbe!H zt>ad%XN$pSY39Q;Q|cRd7N-u?AHe*Tp`LMj{jR+MX^3kt?_7J8+OXc!P^lcm0>@sz zyw@CiYbWhJ4I%2U<8>kJt!KyCfL2=Eh9-8#&A0%Iu!?V_%{A~I!$}C(M$_&(*L>hO z2Urjpb2=d2q2d`FXO)TfI$g{A-EpP2tEOrnD(Y(v@xySe8}*9)$Nny7{3*^gOqqM5 z49|CY+k7!!a35mz4P$6tFRy2=AN9$lV>qkb=WPqb0yXdcP@IP*`=RQQbcfRqS& z+rqK%xCGaR$0dSBd$=O7yCS^XJX?JVUC|!p`|u2TtH!V_KIG$6IaZi8H<)ASj}Q4N zSNI&5qssTs;KFrt)N-u5<8DNs#n8|k^$;!s4&XCClsffiFf0?)$(cRumBEPnxbrGA zrJQ|XH9D|>V-uG0Z5NhtT)`b##k+OPqD!zG!@N(T4{6R^&Yk)^$6ai{Zp3QbiUDTu zg}9gF1Bl~MwpGtyEuQ1*-(Wrdf+7438}Jt2y@QMK0ehSX8kb0^dC3ftHcc#j~=Y%H5sOcFaI z*N|N<@Jl60tPheT*P=<_YYTyIkX+gD*euba0&M2m)#EB=_tnJa8Ul1JKXn~9Zxo%l zz5sCtlj99zU*HhpiBLx=2{~O7!oMZsbhzUxJJ-p{?1q@{5H?SMBD4Jsd$O97jdF5J z@g%ji5e0wDFHvb|9K+W2^15yPsGo~%RQ%pHC8orDWW^B}RXFvwWW{f^E0UFvw=EKj zOo(uGWLzXU+GfU23CT#z=Ry;pe9;SKPRRr{C56JvbA$k(;lepW_z? zo)o2cO3cO6;xv5M-u4bjPBr|Z5S_{o;c?JW3eky~br5_p}p{S7kt8msU>a3$Vkt$T|H&YSG0-X>J<5UO_x&3lC8T|7-#zK4G^?cV1G zALhyrc>R#*93eXI^T0VKX0c~GMR>$Q;r&>AYIx#&6h0wZahsDIJU%_V1EOQNb_}Cq zxIxyZ-EIt=&ai^bBWQ`2s0P@@heQZvw93n}ArI%TvLNt>DAxM>sq1$8dWU_z)4twq ZU+=N6_u1EdEEe)N5W#}+C>}@Ue*i8-@ZJCb literal 0 HcmV?d00001 diff --git a/defects/pulsar/unit/PulsarTest.java b/defects/pulsar/unit/PulsarTest.java new file mode 100644 index 000000000..33a0a9f6c --- /dev/null +++ b/defects/pulsar/unit/PulsarTest.java @@ -0,0 +1,267 @@ +import java.util.*; + +/** + * CWE-407 unit tests for Apache Pulsar. + * + * Three defects measured: + * pulsar-0001: ModularLoadManagerImpl.reapDeadBrokerPreallocations() + * aliveBrokers is List from listLocks(); O(B^2) lookup + * pulsar-0002: PersistentTopic.removeOrphanReplicationCursors() and + * checkReplicationStatus(); configuredClusters is List + * and .contains() is called inside a loop over cursors/replicators + * pulsar-0003: NamespacesBase.internalGetTopicHashPositionsAsync(); + * allTopicsInThisBundle is List 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 knownBrokers, List aliveBrokers) { + // Simulates: for each known broker, scan aliveBrokers list for membership + long ops = 0; + List 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 knownBrokers, List aliveBrokers) { + Set aliveSet = new HashSet<>(aliveBrokers); + long ops = 0; + List 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 knownBrokers = new ArrayList<>(); + List 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 replicationClusters = topicPolicies.getReplicationClusters().get(); + // for (ManagedCursor cursor : ledger.getCursors()) { + // if (!replicationClusters.contains(remoteCluster)) { ... } // O(C) + // } + // + // Pattern (checkReplicationStatus): + // List configuredClusters = topicPolicies.getReplicationClusters().get(); + // replicators.forEach((cluster, replicator) -> { + // if (!configuredClusters.contains(cluster)) { ... } // O(C) + // }); + // ----------------------------------------------------------------------- + + static long defectReplicationCursorScan(List replicationClusters, List cursors) { + long removes = 0; + for (String cursor : cursors) { + // simulates: if (!replicationClusters.contains(remoteCluster)) + if (!replicationClusters.contains(cursor)) { + removes++; + } + } + return removes; + } + + static long fixedReplicationCursorScan(List replicationClusters, List cursors) { + Set 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 replicationClusters = new ArrayList<>(); + List 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 allTopicsInThisBundle = getOwnedTopicListForNamespaceBundle(nsBundle); + // for (String topic : topics) { + // if (allTopicsInThisBundle.contains(topicName.toString())) { ... } // O(B) + // } + // ----------------------------------------------------------------------- + + static long defectTopicHashPositions(List allTopicsInBundle, List 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 allTopicsInBundle, List queryTopics) { + Set 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 allTopicsInBundle = new ArrayList<>(); + List 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"); + } +}