You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2023/07/07 14:13:00 UTC
[pulsar] branch branch-2.10 updated: [fix][broker] Topic policy can not be work well if replay policy message has any exception. (#20613)
This is an automated email from the ASF dual-hosted git repository.
xiangying 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 0a69f661314 [fix][broker] Topic policy can not be work well if replay policy message has any exception. (#20613)
0a69f661314 is described below
commit 0a69f661314c7f79c254afacfb439c559949c838
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Fri Jun 30 12:28:32 2023 +0800
[fix][broker] Topic policy can not be work well if replay policy message has any exception. (#20613)
(cherry picked from commit 200fb562dd4437857ccaba3850bd64b0a9a50b3c)
---
.../service/SystemTopicBasedTopicPoliciesService.java | 18 +++++++++++++++---
1 file changed, 15 insertions(+), 3 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 577fc148e09..0d5fc02532d 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
@@ -176,7 +176,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
TopicName topicName = TopicName.get(TopicName.get(msg.getKey()).getPartitionedTopicName());
if (listeners.get(topicName) != null) {
for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
- listener.onUpdate(null);
+ try {
+ listener.onUpdate(null);
+ } catch (Throwable error) {
+ log.error("[{}] call listener error.", topicName, error);
+ }
}
}
return;
@@ -191,7 +195,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
if (listeners.get(topicName) != null) {
TopicPolicies policies = event.getPolicies();
for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
- listener.onUpdate(policies);
+ try {
+ listener.onUpdate(policies);
+ } catch (Throwable error) {
+ log.error("[{}] call listener error.", topicName, error);
+ }
}
}
}
@@ -362,7 +370,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
policiesCache.forEach(((topicName, topicPolicies) -> {
if (listeners.get(topicName) != null) {
for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
- listener.onUpdate(topicPolicies);
+ try {
+ listener.onUpdate(topicPolicies);
+ } catch (Throwable error) {
+ log.error("[{}] call listener error.", topicName, error);
+ }
}
}
}));