You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/05/20 08:23:53 UTC
(pulsar) branch branch-3.3 updated: [fix][broker] Fix potential deadlock when creating partitioned topic (#24313)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new f7708dc4219 [fix][broker] Fix potential deadlock when creating partitioned topic (#24313)
f7708dc4219 is described below
commit f7708dc42197025597e305f15de93b64abd31811
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon May 19 14:55:54 2025 +0800
[fix][broker] Fix potential deadlock when creating partitioned topic (#24313)
### Motivation
There is still a call to the synchronous method in `internalCreatePartitionedTopic`, which can lead to a deadlock under certain conditions.
### Modifications
- Replaced synchronous call to `getNamespacePolicies()` with `getNamespacePoliciesAsync()`.
- Replaced synchronous call to `getTopicPartitionList()` with `getTopicPartitionListAsync()`.
- Updated `internalCreatePartitionedTopic` to propagate the async flow properly.
(cherry picked from commit 7ca8c0e474a39b9ed51a086e311fda96bbfe43c0)
---
.../apache/pulsar/broker/admin/AdminResource.java | 53 ++++++++++++++--------
1 file changed, 34 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index ce69073e055..a70fddfbb8d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -300,6 +300,10 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ /**
+ * @deprecated Use {@link #getNamespacePoliciesAsync(NamespaceName)} instead.
+ */
+ @Deprecated
protected Policies getNamespacePolicies(NamespaceName namespaceName) {
try {
Policies policies = namespaceResources().getPolicies(namespaceName)
@@ -541,6 +545,10 @@ public abstract class AdminResource extends PulsarWebResource {
.listPartitionedTopicsAsync(namespaceName, topicDomain);
}
+ /**
+ * @deprecated Use {@link #getTopicPartitionListAsync()} instead.
+ */
+ @Deprecated
protected List<String> getTopicPartitionList(TopicDomain topicDomain) {
try {
return getPulsarResources().getTopicResources().getExistingPartitions(topicName)
@@ -552,6 +560,10 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ protected CompletableFuture<List<String>> getTopicPartitionListAsync() {
+ return getPulsarResources().getTopicResources().getExistingPartitions(topicName);
+ }
+
protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions,
boolean createLocalTopicOnly) {
internalCreatePartitionedTopic(asyncResponse, numPartitions, createLocalTopicOnly, null);
@@ -571,33 +583,36 @@ public abstract class AdminResource extends PulsarWebResource {
return;
}
validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.CREATE_TOPIC)
- .thenRun(() -> {
- Policies policies = null;
- try {
- policies = getNamespacePolicies(namespaceName);
- } catch (RestException e) {
- if (e.getResponse().getStatus() != Status.NOT_FOUND.getStatusCode()) {
- throw e;
+ .thenCompose((__) -> getNamespacePoliciesAsync(namespaceName).exceptionally(ex -> {
+ Throwable unwrapped = FutureUtil.unwrapCompletionException(ex);
+ if (unwrapped instanceof RestException re) {
+ if (re.getResponse().getStatus() == Status.NOT_FOUND.getStatusCode()) {
+ return null;
}
}
-
+ throw FutureUtil.wrapToCompletionException(ex);
+ }))
+ .thenCompose(policies -> {
int maxTopicsPerNamespace = policies != null && policies.max_topics_per_namespace != null
? policies.max_topics_per_namespace : pulsar().getConfig().getMaxTopicsPerNamespace();
// new create check
if (maxTopicsPerNamespace > 0 && !pulsar().getBrokerService().isSystemTopic(topicName)) {
- List<String> partitionedTopics = getTopicPartitionList(TopicDomain.persistent);
- // exclude created system topic
- long topicsCount =
- partitionedTopics.stream().filter(t ->
- !pulsar().getBrokerService().isSystemTopic(TopicName.get(t))).count();
- if (topicsCount + numPartitions > maxTopicsPerNamespace) {
- log.error("[{}] Failed to create partitioned topic {}, "
- + "exceed maximum number of topics in namespace", clientAppId(), topicName);
- throw new RestException(Status.PRECONDITION_FAILED,
- "Exceed maximum number of topics in namespace.");
- }
+ return getTopicPartitionListAsync().thenAccept(partitionedTopics -> {
+ // exclude created system topic
+ long topicsCount = partitionedTopics.stream()
+ .filter(t -> !pulsar().getBrokerService().isSystemTopic(TopicName.get(t)))
+ .count();
+ if (topicsCount + numPartitions > maxTopicsPerNamespace) {
+ log.error("[{}] Failed to create partitioned topic {}, "
+ + "exceed maximum number of topics in namespace", clientAppId(),
+ topicName);
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "Exceed maximum number of topics in namespace.");
+ }
+ });
}
+ return CompletableFuture.completedFuture(null);
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
.thenAccept(exists -> {