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/07/25 15:01:15 UTC
(pulsar) 03/06: [fix][broker] Check deliverAt before containsMessage in bucket addMessage (#26230)
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 ded36d47257771642a09a65ba4dc8310c52473eb
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Jul 23 17:20:44 2026 +0800
[fix][broker] Check deliverAt before containsMessage in bucket addMessage (#26230)
(cherry picked from commit 094f2702833ac0ddf71beae4df2072048861eaf5)
---
.../bucket/BucketDelayedDeliveryTracker.java | 8 ++--
.../bucket/BucketDelayedDeliveryTrackerTest.java | 48 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 4 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 6cf2876d4b88..37ee442b85a5 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
@@ -367,14 +367,14 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
@Override
public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) {
- if (containsMessage(ledgerId, entryId)) {
- return true;
- }
-
if (deliverAt < 0 || deliverAt <= getCutoffTime()) {
return false;
}
+ if (containsMessage(ledgerId, entryId)) {
+ return true;
+ }
+
boolean existBucket = findImmutableBucket(ledgerId).isPresent();
// Create bucket snapshot
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 cfe7ed370daf..4a73f335cf48 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
@@ -45,6 +45,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
+import lombok.Cleanup;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
@@ -154,6 +155,10 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
new BucketDelayedDeliveryTracker(dispatcher, timer, 100000, clock,
true, bucketSnapshotStorage, 20, TimeUnit.HOURS.toMillis(1), 5, 100)
}};
+ case "testExpiredTrackedMessageReturnsFalse", "testRecoverThenExpireAddMessage" -> new Object[][]{{
+ new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock,
+ true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50)
+ }};
default -> new Object[][]{{
new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock,
true, bucketSnapshotStorage, 1000, TimeUnit.MILLISECONDS.toMillis(100), -1, 50)
@@ -185,6 +190,49 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
tracker.close();
}
+ @Test(dataProvider = "delayedTracker")
+ public void testExpiredTrackedMessageReturnsFalse(BucketDelayedDeliveryTracker tracker) {
+ clockTime.set(1000);
+ assertTrue(tracker.addMessage(1, 1, 2000));
+ assertTrue(tracker.containsMessage(1, 1));
+
+ clockTime.set(2500);
+
+ assertFalse(
+ "Expired tracked message should return false so dispatcher delivers immediately",
+ tracker.addMessage(1, 1, 2000));
+
+ tracker.close();
+ }
+
+ @Test(dataProvider = "delayedTracker")
+ public void testRecoverThenExpireAddMessage(BucketDelayedDeliveryTracker tracker) throws Exception {
+ clockTime.set(0);
+ for (int i = 1; i <= 6; i++) {
+ tracker.addMessage(i, i, i * 1000);
+ }
+
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging || !x.getSnapshotCreateFuture().get().isDone())));
+
+ tracker.close();
+
+ clockTime.set(0);
+ @Cleanup
+ BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker(
+ dispatcher, timer, 100000, clock, true,
+ bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50);
+
+ assertTrue(tracker2.containsMessage(1, 1));
+
+ clockTime.set(10000);
+
+ assertFalse(
+ "Recovered message that is now expired should return false",
+ tracker2.addMessage(1, 1, 1000));
+ }
+
@Test(dataProvider = "delayedTracker", invocationCount = 10)
public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) throws Exception {
for (int i = 1; i <= 100; i++) {