You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/03/14 03:19:53 UTC
[pulsar] branch branch-2.10 updated: [fix][broker] Fix potential exception cause the policy service init fail. (#19746)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 4449814ca5e [fix][broker] Fix potential exception cause the policy service init fail. (#19746)
4449814ca5e is described below
commit 4449814ca5eb2357257923daf18f03c241dbed7b
Author: Jiwei Guo <te...@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 eee57482ee0..577fc148e09 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
@@ -338,20 +338,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()) {