[pulsar] branch master updated: [fix][broker] Fix potential exception cause the policy service init fail. (#19746)

[technoboy]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 4a450aa1091 [fix][broker] Fix potential exception cause the policy 
service init fail. (#19746)
4a450aa1091 is described below

commit 4a450aa1091d15db14e3d69454c3bb89b3ae6d7d
Author: Jiwei Guo <techno...@apache.org>
AuthorDate: Thu Mar 9 16:18:00 2023 +0800

    [fix][broker] Fix potential exception cause the policy service init fail. 
(#19746)
---
 .../service/SystemTopicBasedTopicPoliciesService.java     | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 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 9ec374264e9..9b10055f36f 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
@@ -341,20 +341,19 @@ public class SystemTopicBasedTopicPoliciesService 
implements TopicPoliciesServic
                 return;
             }
             if (hasMore) {
-                reader.readNextAsync().whenComplete((msg, e) -> {
-                    if (e != null) {
-                        log.error("[{}] Failed to read event from the system 
topic.",
-                                reader.getSystemTopic().getTopicName(), e);
-                        future.completeExceptionally(e);
-                        
cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(),
 false);
-                        return;
-                    }
+                reader.readNextAsync().thenAccept(msg -> {
                     refreshTopicPoliciesCache(msg);
                     if (log.isDebugEnabled()) {
                         log.debug("[{}] Loop next event reading for system 
topic.",
                                 
reader.getSystemTopic().getTopicName().getNamespaceObject());
                     }
                     initPolicesCache(reader, future);
+                }).exceptionally(e -> {
+                    log.error("[{}] Failed to read event from the system 
topic.",
+                            reader.getSystemTopic().getTopicName(), e);
+                    future.completeExceptionally(e);
+                    
cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(),
 false);
+                    return null;
                 });
             } else {
                 if (log.isDebugEnabled()) {