You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2025/09/18 09:17:11 UTC
(pulsar) branch branch-3.0 updated: [fix][broker][branch-3.0] Prevent NPE in ownedBundlesCountPerNamespace on first bundle load (#24758)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 6ceb4f8d749 [fix][broker][branch-3.0] Prevent NPE in ownedBundlesCountPerNamespace on first bundle load (#24758)
6ceb4f8d749 is described below
commit 6ceb4f8d74934480c92da0a15ef4cc0ba5e395ce
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Sep 18 17:17:04 2025 +0800
[fix][broker][branch-3.0] Prevent NPE in ownedBundlesCountPerNamespace on first bundle load (#24758)
---
.../service/SystemTopicBasedTopicPoliciesService.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 4ccbc4c07e9..e2f58b684db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -393,14 +393,14 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
if (NamespaceService.isHeartbeatNamespace(namespace)) {
return CompletableFuture.completedFuture(null);
}
- synchronized (this) {
- if (readerCaches.get(namespace) != null) {
- ownedBundlesCountPerNamespace.get(namespace).incrementAndGet();
- return CompletableFuture.completedFuture(null);
- } else {
- return prepareInitPoliciesCacheAsync(namespace);
- }
+ AtomicInteger bundlesCount =
+ ownedBundlesCountPerNamespace.computeIfAbsent(namespace, k -> new AtomicInteger(0));
+ int previousCount = bundlesCount.getAndIncrement();
+ if (previousCount == 0) {
+ // initialize policies cache asynchronously on the first bundle load
+ return prepareInitPoliciesCacheAsync(namespace);
}
+ return CompletableFuture.completedFuture(null);
}
@VisibleForTesting