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 2026/10/20 13:26:58 UTC
(pulsar) 10/17: [fix][broker] Fix partitioned topic auto deletion http request timeout (#24912)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2-cherry-picks-2026-10-20
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6c0a451bebe00cce5eaa29a5c68d7a2d855bc2d7
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>
(cherry picked from commit f122d21006000b9e17e6ba0d1e36517568e4e66a)
---
.../broker/admin/impl/PersistentTopicsBase.java | 83 ++++++++++++----------
.../broker/service/InactiveTopicDeleteTest.java | 3 +
2 files changed, 49 insertions(+), 37 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 aad4141f5919..68c571a7e10c 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
@@ -888,44 +888,53 @@ 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
- if (log.isDebugEnabled()) {
- log.debug("[{}] Partition not found: {}", clientAppId(),
- topicNamePartition);
- }
- future.complete(null);
- } else {
- log.error("[{}] Failed to delete partition {}", clientAppId(),
- topicNamePartition, realCause);
- future.completeExceptionally(realCause);
- }
- } else {
- future.complete(null);
- }
- });
- return future;
- } catch (PulsarServerException ex) {
- log.error("[{}] Failed to get admin client while delete partition {}",
- clientAppId(), topicNamePartition, ex);
- return FutureUtil.failedFuture(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) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found, skipping its deletion: {}", clientAppId(),
+ topicNamePartition);
}
- }).collect(Collectors.toList()));
+ 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.
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Partition not found: {}", clientAppId(), topicNamePartition);
+ }
+ future.complete(null);
+ } else {
+ log.error("[{}] Failed to delete partition {}", clientAppId(), topicNamePartition,
+ realCause);
+ future.completeExceptionally(realCause);
+ }
+ } 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 73c3488495b7..7ec5774ef4fa 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";