[fix][broker] Fix SystemTopicBasedTopicPoliciesService NPE issue
@@ -246,7 +246,6 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
ownedBundlesCountPerNamespace.get(namespace).incrementAndGet();
result.complete(null);
} else {
- ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1));
prepareInitPoliciesCache(namespace, result);
}
}
@@ -258,6 +257,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerCompletableFuture =
createSystemTopicClientWithRetry(namespace);
readerCaches.put(namespace, readerCompletableFuture);
+ ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1));
readerCompletableFuture.thenAccept(reader -> {
initPolicesCache(reader, result);
result.thenRun(() -> readMorePolicies(reader));
