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/29 15:51:59 UTC
(pulsar) branch branch-4.2 updated (1a7fe8bc7c8 -> 9cf8a613f76)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 1a7fe8bc7c8 [fix][sec][branch-4.2] Upgrade Jackson version to 2.18.8 (#26099)
new ba906cbaff5 [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted (#25984)
new 964a723c7e0 [fix][broker] Guard BucketDelayedDeliveryTracker.nextDeliveryTime against empty queues (#26080)
new 9cf8a613f76 [improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bucket/BucketDelayedDeliveryTracker.java | 115 ++++++++-
...tractPersistentDispatcherMultipleConsumers.java | 10 +
.../PersistentDispatcherMultipleConsumers.java | 2 +-
...rsistentDispatcherMultipleConsumersClassic.java | 2 +-
.../bucket/BucketDelayedDeliveryTrackerTest.java | 259 +++++++++++++++++++++
.../PersistentDispatcherTotalBytesTest.java | 78 +++++++
6 files changed, 453 insertions(+), 13 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java
(pulsar) 03/03: [improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9cf8a613f76a8e97678b7961ee63fc5d6e96e447
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Thu Jun 25 03:34:33 2026 +0800
[improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
(cherry picked from commit b2f37560b356a011bcd9bdb7d37d8ce01d95ea98)
---
...tractPersistentDispatcherMultipleConsumers.java | 10 +++
.../PersistentDispatcherMultipleConsumers.java | 2 +-
...rsistentDispatcherMultipleConsumersClassic.java | 2 +-
.../PersistentDispatcherTotalBytesTest.java | 78 ++++++++++++++++++++++
4 files changed, 90 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
index 79d365b9fee..3a63dc65947 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import java.util.List;
import java.util.Map;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
@@ -64,4 +66,12 @@ public abstract class AbstractPersistentDispatcherMultipleConsumers extends Abst
public abstract Map<String, TopicMetricBean> getBucketDelayedIndexStats();
public abstract boolean isClassic();
+
+ static long getTotalBytesSize(List<Entry> entries) {
+ long totalBytesSize = 0;
+ for (int i = 0, entriesSize = entries.size(); i < entriesSize; i++) {
+ totalBytesSize += entries.get(i).getLength();
+ }
+ return totalBytesSize;
+ }
}
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 c855550069e..71e822e6dd6 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
@@ -707,7 +707,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size());
}
- long totalBytesSize = entries.stream().mapToLong(Entry::getLength).sum();
+ long totalBytesSize = getTotalBytesSize(entries);
updatePendingBytesToDispatch(totalBytesSize);
// dispatch messages to a separate thread, but still in order for this subscription
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 59f2ce973da..7c35e626dc4 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
@@ -621,7 +621,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size());
}
- long size = entries.stream().mapToLong(Entry::getLength).sum();
+ long size = getTotalBytesSize(entries);
updatePendingBytesToDispatch(size);
// dispatch messages to a separate thread, but still in order for this subscription
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java
new file mode 100644
index 00000000000..239a698ab01
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.testng.Assert.assertEquals;
+import java.util.ArrayList;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class PersistentDispatcherTotalBytesTest {
+
+ @DataProvider(name = "entryCounts")
+ public Object[][] entryCounts() {
+ return new Object[][] {
+ {0},
+ {1},
+ {32},
+ {1024}
+ };
+ }
+
+ @Test(dataProvider = "entryCounts")
+ public void testGetTotalBytesSize(int entryCount) {
+ EntriesAndExpectedSize entries = entriesWithVaryingPayloadSizes(entryCount);
+ try {
+ assertEquals(AbstractPersistentDispatcherMultipleConsumers.getTotalBytesSize(entries.entries),
+ entries.expectedSize);
+ } finally {
+ entries.release();
+ }
+ }
+
+ private static EntriesAndExpectedSize entriesWithVaryingPayloadSizes(int entryCount) {
+ EntriesAndExpectedSize entries = new EntriesAndExpectedSize(entryCount);
+ for (int i = 0; i < entryCount; i++) {
+ int payloadSize = payloadSize(i);
+ entries.entries.add(EntryImpl.create(1, i, new byte[payloadSize]));
+ entries.expectedSize += payloadSize;
+ }
+ return entries;
+ }
+
+ private static int payloadSize(int index) {
+ return index % 97 == 0 ? 4096 : 1 + ((index * 31) & 1023);
+ }
+
+ private static final class EntriesAndExpectedSize {
+ private final ArrayList<Entry> entries;
+ private long expectedSize;
+
+ private EntriesAndExpectedSize(int entryCount) {
+ this.entries = new ArrayList<>(entryCount);
+ }
+
+ private void release() {
+ entries.forEach(Entry::release);
+ }
+ }
+}
(pulsar) 02/03: [fix][broker] Guard BucketDelayedDeliveryTracker.nextDeliveryTime against empty queues (#26080)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 964a723c7e07a3a07c93e01c819e08326ba89ac4
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Wed Jun 24 18:02:51 2026 +0800
[fix][broker] Guard BucketDelayedDeliveryTracker.nextDeliveryTime against empty queues (#26080)
(cherry picked from commit 51e2b9a6bc9f739d3fc33b24821df15c588d4055)
---
.../bucket/BucketDelayedDeliveryTracker.java | 5 +++
.../bucket/BucketDelayedDeliveryTrackerTest.java | 46 ++++++++++++++++++++++
2 files changed, 51 insertions(+)
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 7a1b95a3a4a..4b6121fd3e4 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
@@ -593,6 +593,11 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
return sharedBucketPriorityQueue.peekN1();
} else if (sharedBucketPriorityQueue.isEmpty() && !lastMutableBucket.isEmpty()) {
return lastMutableBucket.nextDeliveryTime();
+ } else if (lastMutableBucket.isEmpty() && sharedBucketPriorityQueue.isEmpty()) {
+ // numberDelayedMessages can be > 0 while both queues are empty (e.g. remaining
+ // messages live in not-yet-loaded snapshot segments). Returning Long.MAX_VALUE
+ // signals "no imminent delivery" without throwing on the empty queues.
+ return Long.MAX_VALUE;
}
long timestamp = lastMutableBucket.nextDeliveryTime();
long bucketTimestamp = sharedBucketPriorityQueue.peekN1();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
index ee0ec0d9294..4becc00ab3f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
@@ -676,4 +676,50 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
ts.close();
}
+
+ @Test
+ public void testGetScheduledMessagesWhenAllOrphaned() throws Exception {
+ // Reproduces IAE in nextDeliveryTime: when every delayed message lies below the
+ // mark-delete position, the filter in getScheduledMessages pops the in-memory
+ // messages without returning them. If the immutable bucket has additional messages
+ // still in storage (later snapshot segments), numberDelayedMessages stays > 0
+ // while both the mutable bucket and the shared priority queue are empty.
+ // The trailing updateTimer -> nextDeliveryTime must not throw.
+ long firstLedgerId = 50L;
+ TrackerWithStorage ts = createTrackerWithMockLedger(firstLedgerId, 50);
+
+ // Five delayed messages on the same orphaned ledger (ledgerId < firstLedgerId).
+ // They share a mutable bucket because seal requires a strictly greater ledgerId.
+ // Timestamps are 100ms apart so each lands in its own snapshot segment
+ // (timeStep=10ms); only the first segment is loaded into the shared queue at seal.
+ for (int i = 1; i <= 5; i++) {
+ ts.tracker.addMessage(1, i, i * 100);
+ }
+ // A new orphaned ledgerId triggers the seal, producing immutable bucket [1..1]
+ // with 5 messages across 5 segments; shared queue holds just the first segment.
+ ts.tracker.addMessage(2, 1, 600);
+
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ // In strict deliver-at mode getCutoffTime() is just clock.millis(), so advancing the
+ // clock past the trigger message's deliverAt (600) is enough for
+ // moveScheduledMessageToSharedQueue to flush the mutable bucket into the shared queue.
+ ts.clockTime.set(700);
+
+ // Both queues end up empty (filter pops the two in-memory messages), but
+ // numberDelayedMessages is still 4 (segments 2..5 remain in storage).
+ NavigableSet<Position> scheduledMessages = ts.tracker.getScheduledMessages(10);
+ assertTrue(scheduledMessages.isEmpty(),
+ "Orphaned messages should be filtered out, not returned");
+ assertTrue(ts.tracker.getNumberOfDelayedMessages() > 0,
+ "Remaining storage-only messages should keep the counter > 0");
+
+ // hasMessageAvailable calls nextDeliveryTime while numberDelayedMessages > 0;
+ // it must not throw IAE.
+ assertFalse(ts.tracker.hasMessageAvailable());
+
+ ts.close();
+ }
}
(pulsar) 01/03: [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted (#25984)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ba906cbaff5fb449f38cb4fd8e7ef765310d535f
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Tue Jun 23 15:26:13 2026 +0800
[improve][broker] Trim orphaned bucket snapshots when ledgers are deleted (#25984)
(cherry picked from commit e45b425f927c63b4de4093ee6f9f0218b188b701)
---
.../bucket/BucketDelayedDeliveryTracker.java | 110 +++++++++--
.../bucket/BucketDelayedDeliveryTrackerTest.java | 213 +++++++++++++++++++++
2 files changed, 312 insertions(+), 11 deletions(-)
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 69edf70dad7..7a1b95a3a4a 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
@@ -38,6 +38,7 @@ import java.util.NavigableSet;
import java.util.Optional;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -48,6 +49,7 @@ import javax.annotation.concurrent.ThreadSafe;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.commons.collections4.CollectionUtils;
@@ -112,6 +114,8 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
private CompletableFuture<Void> pendingLoad = null;
+ private volatile CompletableFuture<Void> trimFuture;
+
public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Timer timer, long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict,
@@ -387,8 +391,15 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime);
lastMutableBucket.resetLastMutableBucketRange();
- if (maxNumBuckets > 0 && immutableBuckets.asMapOfRanges().size() > maxNumBuckets) {
- asyncMergeBucketSnapshot();
+ if (maxNumBuckets > 0 && immutableBuckets.asMapOfRanges().size() > maxNumBuckets
+ && (trimFuture == null || trimFuture.isDone())) {
+ trimFuture = asyncTrimImmutableBuckets()
+ .thenCompose(ignore -> asyncMergeBucketSnapshot())
+ .whenComplete((ignore, t) -> {
+ if (t != null) {
+ log.warn("Failed to trim or merge bucket snapshots", t);
+ }
+ });
}
}
@@ -452,6 +463,10 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot() {
List<ImmutableBucket> immutableBucketList = immutableBuckets.asMapOfRanges().values().stream().toList();
+ if (maxNumBuckets <= 0 || immutableBucketList.size() <= maxNumBuckets) {
+ return CompletableFuture.completedFuture(null);
+ }
+
List<ImmutableBucket> toBeMergeImmutableBuckets = selectMergedBuckets(immutableBucketList, MAX_MERGE_NUM);
if (toBeMergeImmutableBuckets.isEmpty()) {
@@ -605,6 +620,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
}
long cutoffTime = getCutoffTime();
+ Long firstLiveLedgerId = firstActiveLedgerId();
lastMutableBucket.moveScheduledMessageToSharedQueue(cutoffTime, sharedBucketPriorityQueue);
@@ -613,13 +629,19 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
while (n > 0 && !sharedBucketPriorityQueue.isEmpty()) {
long timestamp = sharedBucketPriorityQueue.peekN1();
+ long ledgerId = sharedBucketPriorityQueue.peekN2();
+ long entryId = sharedBucketPriorityQueue.peekN3();
+ if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
+ sharedBucketPriorityQueue.pop();
+ if (removeIndexBit(ledgerId, entryId)) {
+ numberDelayedMessages.decrementAndGet();
+ }
+ continue;
+ }
if (timestamp > cutoffTime) {
break;
}
- long ledgerId = sharedBucketPriorityQueue.peekN2();
- long entryId = sharedBucketPriorityQueue.peekN3();
-
SnapshotKey snapshotKey = new SnapshotKey(ledgerId, entryId);
ImmutableBucket bucket = snapshotSegmentLastIndexMap.get(snapshotKey);
@@ -719,12 +741,26 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
@Override
public synchronized CompletableFuture<Void> clear() {
- CompletableFuture<Void> future = cleanImmutableBuckets();
- sharedBucketPriorityQueue.clear();
- lastMutableBucket.clear();
- snapshotSegmentLastIndexMap.clear();
- numberDelayedMessages.set(0);
- return future;
+ // Wait for any in-flight trim+merge to settle, then clear.
+ // Reuse trimFuture to block new triggers until the clear chain completes.
+ CompletableFuture<Void> before = trimFuture != null && !trimFuture.isDone()
+ ? trimFuture : CompletableFuture.completedFuture(null);
+ trimFuture = before
+ .exceptionally(t -> {
+ log.warn("Trim/merge buckets failed, but still clear", t);
+ return null;
+ })
+ .thenCompose(__ -> {
+ synchronized (BucketDelayedDeliveryTracker.this) {
+ CompletableFuture<Void> future = cleanImmutableBuckets();
+ sharedBucketPriorityQueue.clear();
+ lastMutableBucket.clear();
+ snapshotSegmentLastIndexMap.clear();
+ numberDelayedMessages.set(0);
+ return future;
+ }
+ });
+ return trimFuture;
}
@Override
@@ -791,4 +827,56 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
stats.recordBucketSnapshotSizeBytes(totalSnapshotLength.longValue());
return stats.genTopicMetricMap();
}
+
+ /**
+ * Delete orphaned bucket snapshots whose ledger range lies entirely before the earliest
+ * surviving ledger. Buckets are deleted sequentially; the chain stops on first failure
+ * to avoid wasted work when storage is unavailable.
+ */
+ private synchronized CompletableFuture<Void> asyncTrimImmutableBuckets() {
+ Long firstLedgerId = firstActiveLedgerId();
+ if (null == firstLedgerId) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ManagedLedger ledger = dispatcher.getCursor().getManagedLedger();
+
+ Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets =
+ new HashMap<>(immutableBuckets.subRangeMap(Range.lessThan(firstLedgerId)).asMapOfRanges());
+
+ if (toBeDeletedBuckets.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ String ledgerName = ledger.getName();
+ CompletableFuture<Void> chain = CompletableFuture.completedFuture(null);
+ for (Map.Entry<Range<Long>, ImmutableBucket> entry : toBeDeletedBuckets.entrySet()) {
+ chain = chain.thenCompose(__ ->
+ deleteBucketSnapshot(ledgerName, entry.getKey(), entry.getValue()));
+ }
+ return chain;
+ }
+
+ private CompletableFuture<Void> deleteBucketSnapshot(String ledgerName,
+ Range<Long> range, ImmutableBucket bucket) {
+ return bucket.asyncDeleteBucketSnapshot(stats)
+ .handle((__, t) -> {
+ if (t != null) {
+ log.warn("Failed to delete bucket snapshot, LedgerName: {}, BucketKey: {}",
+ ledgerName, bucket.bucketKey());
+ throw new CompletionException(t);
+ }
+ synchronized (this) {
+ snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
+ immutableBuckets.remove(range);
+ numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
+ }
+ return null;
+ });
+ }
+
+ private Long firstActiveLedgerId() {
+ ManagedCursor cursor = dispatcher.getCursor();
+ Position mdp = cursor.getMarkDeletedPosition();
+ return mdp == null ? null : mdp.getLedgerId();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
index 6ff98fa7f70..ee0ec0d9294 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
@@ -40,13 +40,16 @@ import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo.LedgerInfo;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.pulsar.broker.delayed.AbstractDeliveryTrackerTest;
import org.apache.pulsar.broker.delayed.MockBucketSnapshotStorage;
@@ -463,4 +466,214 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
tracker.close();
}
+
+ private static class TrackerWithStorage {
+ final BucketDelayedDeliveryTracker tracker;
+ final MockBucketSnapshotStorage storage;
+ final AtomicLong clockTime;
+
+ TrackerWithStorage(BucketDelayedDeliveryTracker tracker, MockBucketSnapshotStorage storage,
+ AtomicLong clockTime) {
+ this.tracker = tracker;
+ this.storage = storage;
+ this.clockTime = clockTime;
+ }
+
+ void close() throws Exception {
+ tracker.close();
+ storage.close();
+ }
+ }
+
+ private static class BlockingDeleteStorage extends MockBucketSnapshotStorage {
+ final CompletableFuture<Void> firstDeleteFuture = new CompletableFuture<>();
+ final AtomicLong deleteCalls = new AtomicLong();
+
+ @Override
+ public CompletableFuture<Void> deleteBucketSnapshot(long bucketId) {
+ if (deleteCalls.incrementAndGet() <= 4) {
+ return firstDeleteFuture;
+ }
+ return super.deleteBucketSnapshot(bucketId);
+ }
+ }
+
+ private TrackerWithStorage createTrackerWithMockLedger(long firstLedgerId, int maxNumBuckets)
+ throws Exception {
+ return createTrackerWithMockLedger(firstLedgerId, maxNumBuckets, new MockBucketSnapshotStorage());
+ }
+
+ private TrackerWithStorage createTrackerWithMockLedger(long firstLedgerId, int maxNumBuckets,
+ MockBucketSnapshotStorage storage)
+ throws Exception {
+ storage.start();
+
+ ManagedLedger mockLedger = mock(ManagedLedger.class);
+ NavigableMap<Long, LedgerInfo> ledgerInfo = new TreeMap<>();
+ ledgerInfo.put(firstLedgerId, mock(LedgerInfo.class));
+ when(mockLedger.getLedgersInfo()).thenReturn(ledgerInfo);
+ when(mockLedger.getName()).thenReturn("test-ledger");
+
+ ManagedCursor mockCursor = new MockManagedCursor("test-cursor") {
+ @Override
+ public ManagedLedger getManagedLedger() {
+ return mockLedger;
+ }
+
+ @Override
+ public Position getMarkDeletedPosition() {
+ return PositionFactory.create(firstLedgerId, -1);
+ }
+ };
+
+ AbstractPersistentDispatcherMultipleConsumers disp =
+ mock(AbstractPersistentDispatcherMultipleConsumers.class);
+ Clock mockClock = mock(Clock.class);
+ AtomicLong mockClockTime = new AtomicLong();
+ when(mockClock.millis()).then(x -> mockClockTime.get());
+ doReturn(mockCursor).when(disp).getCursor();
+ doReturn("persistent://public/default/testDelay" + " / " + mockCursor.getName()).when(disp).getName();
+
+ BucketDelayedDeliveryTracker tracker = new BucketDelayedDeliveryTracker(disp, mock(Timer.class),
+ 100000, mockClock, true, storage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, maxNumBuckets);
+ return new TrackerWithStorage(tracker, storage, mockClockTime);
+ }
+
+ @Test
+ public void testTrimRemovesOrphanedBuckets() throws Exception {
+ long firstLedgerId = 31L;
+ int messageCount = 36;
+ TrackerWithStorage ts = createTrackerWithMockLedger(firstLedgerId, 5);
+
+ for (int i = 1; i <= messageCount; i++) {
+ ts.tracker.addMessage(i, i, i * 10);
+ }
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ int bucketCount = ts.tracker.getImmutableBuckets().asMapOfRanges().size();
+ assertTrue(bucketCount <= 5,
+ "Bucket count " + bucketCount + " should be <= maxNumBuckets=5 after trim+merge");
+
+ ts.tracker.getImmutableBuckets().asMapOfRanges().forEach((range, bucket) ->
+ assertTrue(range.lowerEndpoint() >= firstLedgerId,
+ "Remaining bucket range " + range + " should be >= " + firstLedgerId));
+
+ long messagesAfterTrim = ts.tracker.getNumberOfDelayedMessages();
+ ts.clockTime.set(messageCount * 10);
+ NavigableSet<Position> scheduledMessages = ts.tracker.getScheduledMessages(1);
+ assertTrue(scheduledMessages.stream().noneMatch(position -> position.getLedgerId() < firstLedgerId),
+ "Trimmed ledgers should not be returned from the loaded shared queue");
+ assertEquals(ts.tracker.getNumberOfDelayedMessages(), messagesAfterTrim - scheduledMessages.size());
+
+ ts.close();
+ }
+
+ @Test
+ public void testTrimHandlesDeleteFailure() throws Exception {
+ long firstLedgerId = 50L;
+ int messageCount = 31;
+ TrackerWithStorage ts = createTrackerWithMockLedger(firstLedgerId, 5);
+
+ // MaxRetryTimes=3 means the first trim delete attempt plus 3 retries = 4 exceptions consumed.
+ for (int i = 0; i < 4; i++) {
+ ts.storage.injectDeleteException(
+ new BucketSnapshotPersistenceException("Delete failed"));
+ }
+
+ for (int i = 1; i <= messageCount; i++) {
+ ts.tracker.addMessage(i, i, i * 10);
+ }
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(ts.storage.deleteExceptionQueue.isEmpty(),
+ "Delete exception should have been consumed"));
+
+ // Trim failed on the first orphaned bucket; the sequential chain stopped, so all
+ // 6 orphaned buckets remain in immutableBuckets.
+ assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().size() > 0,
+ "Orphaned buckets should remain when trim delete fails");
+ ts.tracker.getImmutableBuckets().asMapOfRanges().forEach((range, bucket) ->
+ assertTrue(range.upperEndpoint() < firstLedgerId,
+ "Remaining bucket " + range + " should be an orphaned bucket"));
+
+ // numberDelayedMessages is unchanged because failed deletes do not decrement the count.
+ assertEquals(ts.tracker.getNumberOfDelayedMessages(), messageCount);
+
+ ts.close();
+ }
+
+ @Test
+ public void testClearRunsAfterInFlightTrimFailure() throws Exception {
+ long firstLedgerId = 50L;
+ int messageCount = 31;
+ BlockingDeleteStorage storage = new BlockingDeleteStorage();
+ TrackerWithStorage ts = createTrackerWithMockLedger(firstLedgerId, 5, storage);
+
+ for (int i = 1; i <= messageCount; i++) {
+ ts.tracker.addMessage(i, i, i * 10);
+ }
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(storage.deleteCalls.get() > 0, "Trim delete should be in flight"));
+
+ CompletableFuture<Void> clearFuture = ts.tracker.clear();
+ storage.firstDeleteFuture.completeExceptionally(new BucketSnapshotPersistenceException("Delete failed"));
+
+ clearFuture.get(1, TimeUnit.MINUTES);
+ assertEquals(ts.tracker.getNumberOfDelayedMessages(), 0);
+ assertEquals(ts.tracker.getImmutableBuckets().asMapOfRanges().size(), 0);
+ assertEquals(ts.tracker.getLastMutableBucket().size(), 0);
+ assertEquals(ts.tracker.getSharedBucketPriorityQueue().size(), 0);
+
+ ts.close();
+ }
+
+ @Test
+ public void testTrimWithNoOrphanedBuckets() throws Exception {
+ TrackerWithStorage ts = createTrackerWithMockLedger(0L, 5);
+
+ for (int i = 1; i <= 31; i++) {
+ ts.tracker.addMessage(i, i, i * 10);
+ }
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ int bucketCount = ts.tracker.getImmutableBuckets().asMapOfRanges().size();
+ assertTrue(bucketCount <= 5,
+ "Bucket count " + bucketCount + " should be <= maxNumBuckets=5");
+ assertTrue(bucketCount > 0, "Should have at least one bucket after merge");
+
+ ts.close();
+ }
+
+ @Test
+ public void testMergeEarlyReturnWhenWithinLimit() throws Exception {
+ TrackerWithStorage ts = createTrackerWithMockLedger(0L, 50);
+
+ for (int i = 1; i <= 30; i++) {
+ ts.tracker.addMessage(i, i, i * 10);
+ }
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ int bucketCount = ts.tracker.getImmutableBuckets().asMapOfRanges().size();
+ assertTrue(bucketCount < 50,
+ "Bucket count " + bucketCount + " should be well below maxNumBuckets=50");
+
+ long msgsBefore = ts.tracker.getNumberOfDelayedMessages();
+ ts.tracker.addMessage(200, 200, 200 * 10);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging)));
+
+ assertEquals(ts.tracker.getNumberOfDelayedMessages(), msgsBefore + 1);
+
+ ts.close();
+ }
}