Skip to content

[fix][broker] Fix bucket snapshot trim and segment loading #26280

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
nodece wants to merge 4 commits into
base: master
Choose a base branch
from
+262 −48
Open

[fix][broker] Fix bucket snapshot trim and segment loading #26280

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
05844ff
[fix][broker] Fix bucket snapshot trim and segment loading
nodece Aug 6, 2026
16178cf
Fix test
nodece Aug 6, 2026
047df2b
Fix test
nodece Aug 6, 2026
ef88cfd
Address comment
nodece Aug 7, 2026
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
58 changes: 34 additions & 24 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 @@ -84,8 +84,6 @@ public static record SnapshotKey(long ledgerId, long entryId) {}

static final int AsyncOperationTimeoutSeconds = 60;

private static final Long INVALID_BUCKET_ID = -1L;

private static final int MAX_MERGE_NUM = 4;

private final long minIndexCountPerBucket;
Expand Down Expand Up @@ -354,7 +352,7 @@ private void afterCreateImmutableBucket(Pair<ImmutableBucket, DelayedIndex> immu
immutableBucket);

immutableBucket.getSnapshotCreateFuture().ifPresent(createFuture -> {
CompletableFuture<Long> future = createFuture.handle((bucketId, ex) -> {
CompletableFuture<Long> future = createFuture.whenComplete((bucketId, ex) -> {
if (ex == null) {
immutableBucket.setSnapshotSegments(null);
immutableBucket.asyncUpdateSnapshotLength()
Expand All @@ -369,8 +367,7 @@ private void afterCreateImmutableBucket(Pair<ImmutableBucket, DelayedIndex> immu

stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.create,
System.currentTimeMillis() - startTime);

return bucketId;
return;
}

log.error()
Expand All @@ -397,10 +394,9 @@ private void afterCreateImmutableBucket(Pair<ImmutableBucket, DelayedIndex> immu
snapshotSegmentLastIndexMap.remove(
new SnapshotKey(lastDelayedIndex.getLedgerId(), lastDelayedIndex.getEntryId()));
}
return INVALID_BUCKET_ID;
});
immutableBucket.setSnapshotCreateFuture(future);
});
});
}
}

Expand Down Expand Up @@ -559,16 +555,12 @@ private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot(List<Immut
buckets.stream().map(bucket -> bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE))
.toList();

return FutureUtil.waitForAll(createFutures).thenCompose(bucketId -> {
if (createFutures.stream().anyMatch(future -> INVALID_BUCKET_ID.equals(future.join()))) {
return FutureUtil.failedFuture(new RuntimeException("Can't merge buckets due to bucket create failed"));
}

return FutureUtil.waitForAll(createFutures).thenCompose(__ -> {
List<CompletableFuture<List<SnapshotSegment>>> getAllSnapshotFutures =
buckets.stream().map(ImmutableBucket::getAllSnapshotSegments).toList();

return FutureUtil.waitForAll(getAllSnapshotFutures)
.thenApply(__ -> {
.thenApply(ignore -> {
return CombinedSegmentDelayedIndexQueue.wrap(
getAllSnapshotFutures.stream().map(CompletableFuture::join).toList());
})
Expand Down Expand Up @@ -675,21 +667,20 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)

while (n > 0 && !sharedBucketPriorityQueue.isEmpty()) {
long timestamp = sharedBucketPriorityQueue.peekN1();
long ledgerId = sharedBucketPriorityQueue.peekN2();
long entryId = sharedBucketPriorityQueue.peekN3();
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
sharedBucketPriorityQueue.pop();
removeIndexBit(ledgerId, entryId);
continue;
}
if (timestamp > cutoffTime) {
break;
}

SnapshotKey snapshotKey = new SnapshotKey(ledgerId, entryId);
long ledgerId = sharedBucketPriorityQueue.peekN2();
long entryId = sharedBucketPriorityQueue.peekN3();

boolean orphaned = firstLiveLedgerId != null && ledgerId < firstLiveLedgerId;
SnapshotKey snapshotKey = new SnapshotKey(ledgerId, entryId);
ImmutableBucket bucket = snapshotSegmentLastIndexMap.get(snapshotKey);
if (bucket != null && immutableBuckets.asMapOfRanges().containsValue(bucket)) {
boolean shouldLoadNextSegment = bucket != null
&& immutableBuckets.asMapOfRanges().containsValue(bucket)
&& (!orphaned || bucket.getEndLedgerId() >= firstLiveLedgerId);
if (shouldLoadNextSegment) {
// All message of current snapshot segment are scheduled, try load next snapshot segment
if (bucket.merging) {
log.info()
Expand Down Expand Up @@ -765,6 +756,12 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
}

sharedBucketPriorityQueue.pop();

if (orphaned) {
removeIndexBit(ledgerId, entryId);
continue;
}

// Dedup: queue may carry the same position twice (initial seal + merge); only the
// first delivery of each position decrements the counter via removeIndexBit.
if (removeIndexBit(ledgerId, entryId)) {
Expand Down Expand Up @@ -901,7 +898,12 @@ private synchronized CompletableFuture<Void> asyncTrimImmutableBuckets() {
}

private CompletableFuture<Void> deleteBucketSnapshot(String ledgerName,
Range<Long> range, ImmutableBucket bucket) {
Range<Long> range, ImmutableBucket bucket) {
synchronized (this) {
if (!isCurrentBucket(range, bucket)) {
return CompletableFuture.completedFuture(null);
}
}
return bucket.asyncDeleteBucketSnapshot(stats)
.handle((__, t) -> {
if (t != null) {
Expand All @@ -911,6 +913,9 @@ private CompletableFuture<Void> deleteBucketSnapshot(String ledgerName,
throw new CompletionException(t);
}
synchronized (this) {
if (!isCurrentBucket(range, bucket)) {
return null;
}
snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
removeBucket(range);
bucket.getDelayedIndexBitMap().forEach((ledgerId, bitmap) ->
Expand All @@ -920,6 +925,11 @@ private CompletableFuture<Void> deleteBucketSnapshot(String ledgerName,
});
}

private boolean isCurrentBucket(Range<Long> range, ImmutableBucket expectedBucket) {
ImmutableBucket currentBucket = immutableBuckets.asMapOfRanges().get(range);
return currentBucket == expectedBucket;
}

private Long firstActiveLedgerId() {
ManagedCursor cursor = context.getCursor();
Position mdp = cursor.getMarkDeletedPosition();
Expand All @@ -930,7 +940,7 @@ private void putBucket(Range<Long> range, ImmutableBucket bucket) {
long removedLength = immutableBuckets.subRangeMap(range).asMapOfRanges().values().stream()
.mapToLong(ImmutableBucket::getSnapshotLength)
.sum();

immutableBuckets.put(range, bucket);
immutableBuckets.put(range, bucket);
bucketsCount.set(immutableBuckets.asMapOfRanges().size());
totalSnapshotLengthBytes.addAndGet(bucket.getSnapshotLength() - removedLength);
Expand Down
69 changes: 46 additions & 23 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,32 +319,55 @@ CompletableFuture<Void> asyncDeleteBucketSnapshot(BucketDelayedMessageIndexStats
long deleteStartTime = System.currentTimeMillis();
stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.delete);
String bucketKey = bucketKey();
long bucketId = getAndUpdateBucketId();
CompletableFuture<Optional<Long>> bucketIdFuture = resolveBucketIdForDelete();

return executeWithRetry(() -> ctx.bucketSnapshotStorage().deleteBucketSnapshot(bucketId),
BucketSnapshotPersistenceException.class, MaxRetryTimes)
.whenComplete((__, ex) -> {
if (ex != null) {
log.error()
.attr("dispatcher", ctx.dispatcherName())
.attr("bucketId", bucketId)
.attr("bucketKey", bucketKey)
.exception(ex)
.log("Failed to delete bucket snapshot");
return bucketIdFuture.thenCompose(optionalBucketId -> {
if (optionalBucketId.isEmpty()) {
return CompletableFuture.completedFuture(null);
}
long bucketId = optionalBucketId.get();
return executeWithRetry(() -> ctx.bucketSnapshotStorage().deleteBucketSnapshot(bucketId),
BucketSnapshotPersistenceException.class, MaxRetryTimes)
.whenComplete((__, ex) -> {
if (ex != null) {
log.error()
.attr("dispatcher", ctx.dispatcherName())
.attr("bucketId", bucketId)
.attr("bucketKey", bucketKey)
.exception(ex)
.log("Failed to delete bucket snapshot");

stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.delete);
} else {
log.info()
.attr("dispatcher", ctx.dispatcherName())
.attr("bucketId", bucketId)
.attr("bucketKey", bucketKey)
.log("Delete bucket snapshot finish");
stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.delete);
} else {
log.info()
.attr("dispatcher", ctx.dispatcherName())
.attr("bucketId", bucketId)
.attr("bucketKey", bucketKey)
.log("Delete bucket snapshot finish");

stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.delete,
System.currentTimeMillis() - deleteStartTime);
}
});
}).thenCompose(__ -> removeBucketCursorProperty(bucketKey));
}

stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.delete,
System.currentTimeMillis() - deleteStartTime);
}
})
.thenCompose(__ -> removeBucketCursorProperty(bucketKey));
private CompletableFuture<Optional<Long>> resolveBucketIdForDelete() {
Optional<CompletableFuture<Long>> snapshotCreateFuture = getSnapshotCreateFuture();
if (snapshotCreateFuture.isPresent()) {
return snapshotCreateFuture.get().handle((bucketId, ex) -> {
if (ex != null) {
return Optional.empty();
}
setBucketId(bucketId);
return Optional.of(bucketId);
});
}
try {
return CompletableFuture.completedFuture(Optional.of(getAndUpdateBucketId()));
} catch (Exception e) {
return FutureUtil.failedFuture(e);
}
}

CompletableFuture<Void> clear(BucketDelayedMessageIndexStats stats) {
Expand Down
Loading
Loading