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/06/18 11:23:15 UTC
(pulsar) branch master updated: [fix][broker] Avoid blocking the dispatcher close path on delayed-delivery tracker close (#26053)
This is an automated email from the ASF dual-hosted git repository.
lhotari 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 82074aa80c5 [fix][broker] Avoid blocking the dispatcher close path on delayed-delivery tracker close (#26053)
82074aa80c5 is described below
commit 82074aa80c5ca8d849b640a368c254e38e5163e2
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Jun 18 04:23:09 2026 -0700
[fix][broker] Avoid blocking the dispatcher close path on delayed-delivery tracker close (#26053)
---
.../broker/delayed/DelayedDeliveryTracker.java | 10 ++++++++
.../bucket/BucketDelayedDeliveryTracker.java | 27 ++++++++++++++--------
.../PersistentDispatcherMultipleConsumers.java | 8 ++++---
...rsistentDispatcherMultipleConsumersClassic.java | 8 ++++---
4 files changed, 38 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
index 7c954879fe8..96e7151e7ea 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
@@ -86,6 +86,16 @@ public interface DelayedDeliveryTracker extends AutoCloseable {
*/
void close();
+ /**
+ * Close the subscription tracker and release all resources, completing the returned future once the
+ * tracker has finished closing. Prefer this over {@link #close()} on asynchronous paths (e.g. inside a
+ * {@link CompletableFuture} continuation) so the caller is not blocked.
+ */
+ default CompletableFuture<Void> closeAsync() {
+ close();
+ return CompletableFuture.completedFuture(null);
+ }
+
DelayedDeliveryTracker DISABLE = new DelayedDeliveryTracker() {
@Override
public boolean addMessage(long ledgerId, long entryId, long deliveryAt) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
index 0b0b48b8de3..29228e634ce 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -758,17 +758,26 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
}
@Override
- public synchronized void close() {
- super.close();
- lastMutableBucket.close();
- sharedBucketPriorityQueue.close();
- try {
- List<CompletableFuture<Long>> completableFutures = immutableBuckets.asMapOfRanges().values().stream()
+ public void close() {
+ // Block for AutoCloseable / synchronous callers; asynchronous callers should use closeAsync().
+ closeAsync().join();
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ List<CompletableFuture<Long>> completableFutures;
+ synchronized (this) {
+ super.close();
+ lastMutableBucket.close();
+ sharedBucketPriorityQueue.close();
+ completableFutures = immutableBuckets.asMapOfRanges().values().stream()
.map(bucket -> bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)).toList();
- FutureUtil.waitForAll(completableFutures).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS);
- } catch (Exception e) {
- log.warn().exception(e).log("Failed wait to snapshot generate");
}
+ return FutureUtil.waitForAll(completableFutures)
+ .exceptionally(e -> {
+ log.warn().exception(e).log("Failed wait to snapshot generate");
+ return null;
+ });
}
private CompletableFuture<Void> cleanImmutableBuckets() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 9bcf9153572..15a6071c23c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -616,11 +616,13 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
this.delayedDeliveryTracker = Optional.empty();
}
- delayedDeliveryTracker.ifPresent(DelayedDeliveryTracker::close);
+ CompletableFuture<Void> closeTrackerFuture = delayedDeliveryTracker
+ .map(DelayedDeliveryTracker::closeAsync)
+ .orElseGet(() -> CompletableFuture.completedFuture(null));
dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
- return disconnectConsumers
- ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null);
+ return closeTrackerFuture.thenCompose(__ -> disconnectConsumers
+ ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null));
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
index 276f8c038a6..be1a29ad6bf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
@@ -528,11 +528,13 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
this.delayedDeliveryTracker = Optional.empty();
}
- delayedDeliveryTracker.ifPresent(DelayedDeliveryTracker::close);
+ CompletableFuture<Void> closeTrackerFuture = delayedDeliveryTracker
+ .map(DelayedDeliveryTracker::closeAsync)
+ .orElseGet(() -> CompletableFuture.completedFuture(null));
dispatchRateLimiter.ifPresent(DispatchRateLimiter::close);
- return disconnectConsumers
- ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null);
+ return closeTrackerFuture.thenCompose(__ -> disconnectConsumers
+ ? disconnectAllConsumers(false, assignedBrokerLookupData) : CompletableFuture.completedFuture(null));
}
@Override