Fix TopicPoliciesCacheNotInitException issue.

[Technoboy-]

Motivation

Sometimes, we may get TopicPoliciesCacheNotInitException with below stack trace:

15:45:47.020 [pulsar-web-41-3] INFO  org.eclipse.jetty.server.RequestLog - 10.0.0.42 - - [10/Nov/2021:15:45:47 +0000] "GET /status.html HTTP/1.1" 200 2 "-" "kube-probe/1.19+" 1
15:45:51.221 [pulsar-2-15] ERROR org.apache.pulsar.broker.admin.impl.PersistentTopicsBase - [null] Failed to perform getRetention on topic persistent://public/default/UpdateNodeCharts
java.lang.RuntimeException: org.apache.pulsar.broker.service.BrokerServiceException$TopicPoliciesCacheNotInitException: Topic policies cache have not init.
	at org.apache.pulsar.broker.service.TopicPoliciesService.lambda$getTopicPoliciesAsyncWithRetry$0(TopicPoliciesService.java:84) ~[io.streamnative-pulsar-broker-2.8.1.21.jar:2.8.1.21]
	at org.apache.pulsar.client.util.RetryUtil.executeWithRetry(RetryUtil.java:50) ~[io.streamnative-pulsar-client-original-2.8.1.21.jar:2.8.1.21]
	at org.apache.pulsar.client.util.RetryUtil.lambda$executeWithRetry$1(RetryUtil.java:63) ~[io.streamnative-pulsar-client-original-2.8.1.21.jar:2.8.1.21]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.68.Final.jar:4.1.68.Final]
	at java.lang.Thread.run(Thread.java:829) [?:?]

This is because :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

Lines 298 to 312 in c3da145

	if (hasMore) {
	reader.readNextAsync().whenComplete((msg, e) -> {
	if (e != null) {
	log.error("[{}] Failed to read event from the system topic.",
	reader.getSystemTopic().getTopicName(), ex);
	future.completeExceptionally(e);
	readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
	}
	refreshTopicPoliciesCache(msg);
	if (log.isDebugEnabled()) {
	log.debug("[{}] Loop next event reading for system topic.",
	reader.getSystemTopic().getTopicName().getNamespaceObject());
	}
	initPolicesCache(reader, future);
	});

when reader.readNextAsync() throws exceptions, the msg will be null which will throw NPE without any catch block.

Modifications

• Close reader when reader.readNextAsync() throw exceptions.
• Remove reducant field policyCacheInitMap . we can use readerCaches instead.
• Add retry logic when getTopicPolicies.

Documentation

• [ x ] no-need-doc

[Technoboy-]

/pulsarbot run-failure-checks

[Technoboy-]

/pulsarbot run-failure-checks

[Technoboy-]

/pulsarbot run-failure-checks