Skip to content

[fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures #26260

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
void-ptr974 wants to merge 1 commit into
base: master
Choose a base branch
from
+110 −14
Open

[fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures #26260

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 29 additions & 12 deletions ...r/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,11 @@ public static record SnapshotKey(long ledgerId, long entryId) {}

private volatile CompletableFuture<Void> trimFuture;

@VisibleForTesting
CompletableFuture<Void> getTrimFuture() {
return trimFuture;
}

public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Timer timer, long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict,
Expand Down Expand Up @@ -918,20 +923,32 @@ private synchronized CompletableFuture<Void> asyncTrimImmutableBuckets() {

private CompletableFuture<Void> deleteBucketSnapshot(String ledgerName,
Range<Long> range, ImmutableBucket bucket) {
return bucket.asyncDeleteBucketSnapshot(stats)
.handle((__, t) -> {
if (t != null) {
log.warn().attr("LedgerName", ledgerName)
.attr("BucketKey", bucket.bucketKey())
.log("Failed to delete bucket snapshot");
throw new CompletionException(t);
}
return bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)

@Denovo1998 Denovo1998 Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This ensures the global trimFuture waits for every selected bucket's snapshot-creation future. If any create future remains pending for an extended period, trimFuture.isDone() stays false, which prevents later addMessage() calls from triggering another trim/merge and also chains clear() behind the same future.

Would it be safer to skip buckets still in the CREATING state and retrigger trim when their creation finishes, rather than holding the global trim/merge gate? At minimum, it would be helpful to document the bounded-completion guarantee or add a test for a create future that never completes.

.thenCompose(bucketId -> {
synchronized (this) {
snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
removeBucket(range);
numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
Long firstLedgerId = firstActiveLedgerId();
if (INVALID_BUCKET_ID.equals(bucketId)
|| firstLedgerId == null
|| range.upperEndpoint() >= firstLedgerId
|| immutableBuckets.asMapOfRanges().get(range) != bucket) {
return CompletableFuture.completedFuture(null);
}
}
return null;
return bucket.asyncDeleteBucketSnapshot(stats).thenRun(() -> {

@Denovo1998 Denovo1998 Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The orphan eligibility is validated only before the asynchronous snapshot deletion begins. Once this synchronized block is released, the cursor may be reset or its mark‑deleted position could move backward while asyncDeleteBucketSnapshot() is still running.

In such cases, the completion block only rechecks the range‑to‑bucket identity, so it may still delete bucket state even if range.upperEndpoint() < firstActiveLedgerId() no longer holds. The existing test updates firstLedgerId before the snapshot‑create future is released, but it does not cover modifications after this validation and after storage deletion has started.

Could we add a deletion ownership or version guard that spans the duration of the asynchronous operation, or base orphan cleanup on a monotonic physical‑ledger boundary? Please also include a test that blocks deleteBucketSnapshot(), changes firstLedgerId after deletion begins, and then completes the delete.

synchronized (this) {
if (immutableBuckets.asMapOfRanges().get(range) != bucket) {
return;
}
snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
removeBucket(range);
numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
}
});
}).exceptionally(t -> {
log.warn().attr("LedgerName", ledgerName)
.attr("BucketKey", bucket.bucketKey())
.log("Failed to delete bucket snapshot");
throw new CompletionException(t);
});
}

Expand Down
83 changes: 81 additions & 2 deletions ...c/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@
import org.apache.pulsar.broker.delayed.AbstractDeliveryTrackerTest;
import org.apache.pulsar.broker.delayed.MockBucketSnapshotStorage;
import org.apache.pulsar.broker.delayed.MockManagedCursor;
import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata;
import org.apache.pulsar.broker.delayed.proto.SnapshotSegment;
import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.awaitility.Awaitility;
import org.roaringbitmap.RoaringBitmap;
Expand Down Expand Up @@ -556,6 +558,28 @@ public CompletableFuture<Void> deleteBucketSnapshot(long bucketId) {
}
}

private static class BlockingCreateStorage extends MockBucketSnapshotStorage {
final CompletableFuture<Void> allowCreate = new CompletableFuture<>();
final AtomicLong createCalls = new AtomicLong();
final AtomicLong deleteCalls = new AtomicLong();

@Override
public CompletableFuture<Long> createBucketSnapshot(
SnapshotMetadata snapshotMetadata, List<SnapshotSegment> bucketSnapshotSegments, String bucketKey,
String topicName, String cursorName) {
createCalls.incrementAndGet();
return super.createBucketSnapshot(snapshotMetadata, bucketSnapshotSegments, bucketKey, topicName,
cursorName)
.thenCompose(bucketId -> allowCreate.thenApply(ignored -> bucketId));
}

@Override
public CompletableFuture<Void> deleteBucketSnapshot(long bucketId) {
deleteCalls.incrementAndGet();
return super.deleteBucketSnapshot(bucketId);
}
}

private static class RecordingDeleteStorage extends MockBucketSnapshotStorage {
final AtomicLong deleteCalls = new AtomicLong();

Expand Down Expand Up @@ -586,11 +610,17 @@ private TrackerWithStorage createTrackerWithMockLedger(long firstLedgerId, int m
private TrackerWithStorage createTrackerWithMockLedger(long firstLedgerId, int maxNumBuckets,
MockBucketSnapshotStorage storage)
throws Exception {
return createTrackerWithMockLedger(new AtomicLong(firstLedgerId), maxNumBuckets, storage);
}

private TrackerWithStorage createTrackerWithMockLedger(AtomicLong 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));
ledgerInfo.put(firstLedgerId.get(), mock(LedgerInfo.class));
when(mockLedger.getLedgersInfo()).thenReturn(ledgerInfo);
when(mockLedger.getName()).thenReturn("test-ledger");

Expand All @@ -602,7 +632,7 @@ public ManagedLedger getManagedLedger() {

@Override
public Position getMarkDeletedPosition() {
return PositionFactory.create(firstLedgerId, -1);
return PositionFactory.create(firstLedgerId.get(), -1);
}
};

Expand Down Expand Up @@ -830,6 +860,55 @@ public void testClearRunsAfterInFlightTrimFailure() throws Exception {
ts.close();
}

@Test
public void testTrimWaitsForSnapshotCreation() throws Exception {
BlockingCreateStorage storage = new BlockingCreateStorage();
TrackerWithStorage ts = createTrackerWithMockLedger(50L, 5, storage);
try {
for (int i = 1; i <= 31; i++) {
ts.tracker.addMessage(i, i, i * 10);
}

assertEquals(storage.createCalls.get(), 6L);
assertEquals(storage.deleteCalls.get(), 0L,
"Trim must not delete a snapshot while its creation is in flight");

storage.allowCreate.complete(null);
ts.tracker.getTrimFuture().get(1, TimeUnit.MINUTES);

assertEquals(storage.deleteCalls.get(), 6L);
assertTrue(ts.tracker.getImmutableBuckets().asMapOfRanges().isEmpty());
} finally {
storage.allowCreate.complete(null);
ts.close();
}
}

@Test
public void testTrimRevalidatesBucketAfterSnapshotCreation() throws Exception {
AtomicLong firstLedgerId = new AtomicLong(50L);
BlockingCreateStorage storage = new BlockingCreateStorage();
TrackerWithStorage ts = createTrackerWithMockLedger(firstLedgerId, 5, storage);
try {
for (int i = 1; i <= 31; i++) {
ts.tracker.addMessage(i, i, 1000L);
}

assertEquals(storage.createCalls.get(), 6L);
firstLedgerId.set(0L);
storage.allowCreate.complete(null);
ts.tracker.getTrimFuture().get(1, TimeUnit.MINUTES);

assertEquals(storage.deleteCalls.get(), 0L,
"Trim must revalidate that a bucket is still orphaned after snapshot creation");
assertEquals(ts.tracker.getImmutableBuckets().asMapOfRanges().size(), 6);
assertEquals(ts.tracker.getNumberOfDelayedMessages(), 31L);
} finally {
storage.allowCreate.complete(null);
ts.close();
}
}

@Test
public void testTrimWithNoOrphanedBuckets() throws Exception {
TrackerWithStorage ts = createTrackerWithMockLedger(0L, 5);
Expand Down