You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/17 16:53:08 UTC
(pulsar) branch master updated: [fix][broker] Fix partitioned topic auto deletion http request timeout (#24912)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 f122d2100600 [fix][broker] Fix partitioned topic auto deletion http request timeout (#24912)
f122d2100600 is described below
commit f122d21006000b9e17e6ba0d1e36517568e4e66a
Author: Oneby Wang <89...@qq.com>
AuthorDate: Sun Oct 18 00:53:01 2026 +0800
[fix][broker] Fix partitioned topic auto deletion http request timeout (#24912)
Co-authored-by: oneby-wang <on...@qq.com>
Co-authored-by: Matteo Merli <mm...@apache.org>
---
.../broker/admin/impl/PersistentTopicsBase.java | 83 ++++++++++++----------
.../broker/service/InactiveTopicDeleteTest.java | 3 +
2 files changed, 48 insertions(+), 38 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index ae5b247ab18b..139c106a4f46 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -919,47 +919,54 @@ public class PersistentTopicsBase extends AdminResource {
private CompletableFuture<Void> internalRemovePartitionsTopicNoAutoCreationDisableAsync(int numPartitions,
boolean force) {
- return FutureUtil.waitForAll(IntStream.range(0, numPartitions)
- .mapToObj(i -> {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- CompletableFuture<Void> future = new CompletableFuture<>();
- pulsar().getAdminClient().topics()
- .deleteAsync(topicNamePartition.toString(), force)
- .whenComplete((r, ex) -> {
- if (ex != null) {
- Throwable realCause = FutureUtil.unwrapCompletionException(ex);
- if (realCause instanceof NotFoundException){
- // if the sub-topic is not found, the client might not have called
- // create producer or it might have been deleted earlier,
- // so we ignore the 404 error.
- // For all other exception,
- // we fail the delete partition method even if a single
- // partition is failed to be deleted
- log.debug()
- .attr("topic", topicNamePartition)
- .log("Partition not found");
- future.complete(null);
- } else {
- log.error()
- .attr("partition", topicNamePartition)
- .exceptionMessage(realCause)
- .log("Failed to delete partition");
- future.completeExceptionally(realCause);
- }
- } else {
- future.complete(null);
- }
- });
- return future;
- } catch (PulsarServerException ex) {
+ return FutureUtil.waitForAll(IntStream.range(0, numPartitions).mapToObj(i -> {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ // Skip the admin call for persistent partitions that don't exist. Inactive-topic deletion deletes all
+ // the partitions first and then calls this endpoint through the broker's own admin client, so each
+ // per-partition admin call would queue behind the outer requests in the per-broker admin client
+ // pool. With enough concurrent deletions, that pool is exhausted and every request times out.
+ // See https://github.com/apache/pulsar/issues/24879
+ // Non-persistent topics have no managed ledger to check, and their inactive-topic deletion doesn't
+ // call this endpoint, so they always go through the admin client.
+ CompletableFuture<Boolean> needAdminClientCall = topicName.isPersistent()
+ ? pulsar().getPulsarResources().getTopicResources().persistentTopicExists(topicNamePartition)
+ : CompletableFuture.completedFuture(true);
+ needAdminClientCall.thenCompose(deleteThroughAdminClient -> {
+ if (!deleteThroughAdminClient) {
+ log.debug()
+ .attr("partition", topicNamePartition)
+ .log("Partition not found, skipping its deletion");
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ try {
+ return pulsar().getAdminClient().topics().deleteAsync(topicNamePartition.toString(), force);
+ } catch (PulsarServerException ex) {
+ return FutureUtil.<Void>failedFuture(ex);
+ }
+ }).whenComplete((r, ex) -> {
+ if (ex != null) {
+ Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+ if (realCause instanceof NotFoundException) {
+ // The partition was deleted after the existence check, or the client never created it:
+ // ignore the 404. Any other failure fails the whole partitioned topic deletion.
+ log.debug()
+ .attr("partition", topicNamePartition)
+ .log("Partition not found");
+ future.complete(null);
+ } else {
log.error()
.attr("partition", topicNamePartition)
- .exception(ex)
- .log("Failed to get admin client while delete partition");
- return FutureUtil.failedFuture(ex);
+ .exceptionMessage(realCause)
+ .log("Failed to delete partition");
+ future.completeExceptionally(realCause);
}
- }).collect(Collectors.toList()));
+ } else {
+ future.complete(null);
+ }
+ });
+ return future;
+ }).collect(Collectors.toList()));
}
private CompletableFuture<Void> internalRemovePartitionsAuthenticationPoliciesAsync() {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
index 6826717e7802..da9e016e02c1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
@@ -114,6 +114,9 @@ public class InactiveTopicDeleteTest extends BrokerTestBase {
conf.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
conf.setBrokerDeleteInactivePartitionedTopicMetadataEnabled(true);
conf.setBrokerDeleteInactiveTopicsFrequencySeconds(1);
+ // A single admin client connection per broker: before the fix for #24879, the per-partition deletes
+ // queued behind the broker's own delete-partitioned-topic request and the metadata was never deleted
+ conf.getProperties().put("brokerClient_connectionsPerBroker", 1);
super.baseSetup();
final String topic = "persistent://prop/ns-abc/testDeleteWhenNoSubscriptions";