You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2025/10/09 01:51:01 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Allow intermittent error from topic policies service when loading topics (#24829)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new dd149d4ba4b [fix][broker] Allow intermittent error from topic policies service when loading topics (#24829)
dd149d4ba4b is described below
commit dd149d4ba4b9bc59bbc793fb083abe09252be5bf
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Oct 8 23:42:11 2025 +0800
[fix][broker] Allow intermittent error from topic policies service when loading topics (#24829)
(cherry picked from commit 7b6f9fceeb56a1e85bdd917e97393ab8cb19544b)
---
.../pulsar/broker/service/BrokerService.java | 22 +++++++++++-------
.../pulsar/broker/service/BrokerServiceTest.java | 27 +++++++++++++++++++++-
2 files changed, 40 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index c2fe905822d..8478e7f0510 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1104,14 +1104,7 @@ public class BrokerService implements Closeable {
// The topic level policies are not needed now, but the meaning of calling
// "getTopicPoliciesBypassSystemTopic" will wait for system topic policies initialization.
getTopicPoliciesBypassSystemTopic(topicName, TopicPoliciesService.GetType.LOCAL_ONLY)
- .exceptionally(ex -> {
- final Throwable rc = FutureUtil.unwrapCompletionException(ex);
- final String errorInfo = String.format("Topic creation encountered an exception by initialize"
- + " topic policies service. topic_name=%s error_message=%s", topicName,
- rc.getMessage());
- log.error(errorInfo, rc);
- throw FutureUtil.wrapToCompletionException(new ServiceUnitNotReadyException(errorInfo));
- }).thenCompose(optionalTopicPolicies -> {
+ .thenCompose(optionalTopicPolicies -> {
if (topicName.isPartitioned()) {
final TopicName topicNameEntity = TopicName.get(topicName.getPartitionedTopicName());
return fetchPartitionedTopicMetadataAsync(topicNameEntity)
@@ -1155,7 +1148,20 @@ public class BrokerService implements Closeable {
}
});
}
+ }).exceptionally(e -> {
+ pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture));
+ final Throwable rc = FutureUtil.unwrapCompletionException(e);
+ final String errorInfo = String.format("Topic creation encountered an exception by initialize"
+ + " topic policies service. topic_name=%s error_message=%s", topicName,
+ rc.getMessage());
+ log.error(errorInfo, rc);
+ topicFuture.completeExceptionally(rc);
+ return null;
});
+ }).exceptionally(e -> {
+ pulsar.getExecutor().execute(() -> topics.remove(topicName.toString(), topicFuture));
+ topicFuture.completeExceptionally(e.getCause());
+ return null;
});
return topicFuture;
} else {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 62dcc37f38e..56a08eac209 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -125,6 +125,7 @@ import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
import org.apache.pulsar.common.policies.data.OffloadedReadPriority;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
@@ -148,7 +149,8 @@ public class BrokerServiceTest extends BrokerTestBase {
@Override
protected void setup() throws Exception {
conf.setSystemTopicEnabled(false);
- conf.setTopicLevelPoliciesEnabled(false);
+ conf.setTopicLevelPoliciesEnabled(true);
+ conf.setTopicPoliciesServiceClassName(MockTopicPoliciesService.class.getName());
super.baseSetup();
}
@@ -2036,5 +2038,28 @@ public class BrokerServiceTest extends BrokerTestBase {
retryStrategically((test) -> sync.getProducer() != null, 1000, 10);
assertNotNull(sync.getProducer());
}
+
+ @Test
+ public void testGetTopicWhenTopicPoliciesFail() throws Exception {
+ final var topicName = TopicName.get("prop/ns-abc/test-get-topic-when-topic-policies-fail");
+ MockTopicPoliciesService.FAILED_TOPICS.add(topicName);
+ @Cleanup final var producer = pulsarClient.newProducer().topic(topicName.toString()).create();
+ assertFalse(MockTopicPoliciesService.FAILED_TOPICS.contains(topicName));
+ }
+
+ static class MockTopicPoliciesService extends TopicPoliciesService.TopicPoliciesServiceDisabled {
+
+ static final Set<TopicName> FAILED_TOPICS = ConcurrentHashMap.newKeySet();
+
+ @Override
+ public CompletableFuture<Optional<TopicPolicies>> getTopicPoliciesAsync(TopicName topicName, GetType type) {
+ if (FAILED_TOPICS.contains(topicName)) {
+ // Only fail once
+ FAILED_TOPICS.remove(topicName);
+ return CompletableFuture.failedFuture(new RuntimeException("injected failure for " + topicName));
+ }
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ }
}