[Branch-2.9] Fix SystemTopicBasedTopicPoliciesService NPE issue.

[Technoboy-]

Motivation

When SystemTopicBasedTopicPoliciesService#getTopicPolicies is invoked before SystemTopicBasedTopicPoliciesService#addOwnedNamespaceBundleAsync, it will cause NPE:

Stack trace

java.lang.NullPointerException: null
	at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.addOwnedNamespaceBundleAsync(SystemTopicBasedTopicPoliciesService.java:222) ~[classes/:?]
	at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$1.onLoad(SystemTopicBasedTopicPoliciesService.java:301) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.notifyNamespaceBundleOwnershipListener(NamespaceService.java:1052) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.onNamespaceBundleOwned(NamespaceService.java:1020) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.OwnershipCache.lambda$tryAcquiringOwnership$1(OwnershipCache.java:200) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:642) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073) ~[?:?]
	at org.apache.pulsar.metadata.coordination.impl.LockManagerImpl.lambda$acquireLock$1(LockManagerImpl.java:105) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniRun.tryFire$$$capture(CompletableFuture.java:783) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniRun.tryFire(CompletableFuture.java) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073) ~[?:?]
	at org.apache.pulsar.metadata.coordination.impl.ResourceLockImpl.lambda$acquire$2(ResourceLockImpl.java:127) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniRun.tryFire$$$capture(CompletableFuture.java:783) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniRun.tryFire(CompletableFuture.java) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073) ~[?:?]
	at org.apache.pulsar.metadata.coordination.impl.ResourceLockImpl.lambda$acquireWithNoRevalidation$6(ResourceLockImpl.java:166) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniAccept.tryFire$$$capture(CompletableFuture.java:714) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073) ~[?:?]
	at org.apache.pulsar.metadata.impl.ZKMetadataStore.handlePutResult(ZKMetadataStore.java:214) ~[classes/:?]
	at org.apache.pulsar.metadata.impl.ZKMetadataStore.lambda$batchOperation$6(ZKMetadataStore.java:171) ~[classes/:?]
	at org.apache.pulsar.metadata.impl.PulsarZooKeeperClient$3$1.processResult(PulsarZooKeeperClient.java:490) [classes/:?]
	at org.apache.zookeeper.ClientCnxn$EventThread.processEvent(ClientCnxn.java:722) [zookeeper-3.6.3.jar:3.6.3]
	at org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:563) [zookeeper-3.6.3.jar:3.6.3]

Modifications

1. Move ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1)) in the prepareInitPoliciesCache method.

Documentation

• no-need-doc

[codelipenghui]

If we move this line here, can we ensure the ownedBundlesCount will decrease to 0 after all the bundles of the namespace are unloaded from this broker?

[Technoboy-]

yes. This is only a init operation, we need to keep readerCaches and ownedBundlesCountPerNamespace the data consistent.

[codelipenghui]

Oh, I see. My mistake.