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/22 12:03:44 UTC
(pulsar) 14/17: [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 branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8ca3fd0d890c7ece76cd4c5abe48d44fc7cf1352
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)
(cherry picked from commit 82074aa80c5ca8d849b640a368c254e38e5163e2)
---
.../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 7c954879fe84..96e7151e7ea4 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 9ed304c73312..77c4798232b5 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
@@ -733,17 +733,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("[{}] Failed wait to snapshot generate", dispatcher.getName(), e);
}
+ return FutureUtil.waitForAll(completableFutures)
+ .exceptionally(e -> {
+ log.warn("[{}] Failed wait to snapshot generate", dispatcher.getName(), e);
+ 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 da6fc52b9163..e51f8a3acf8f 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
@@ -621,11 +621,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 e87b789d8b9f..0838b8deab7f 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
@@ -534,11 +534,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