Skip to content

[fix][broker] Manage in-flight bucket work during clear #26401

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 3 commits into
base: master
Choose a base branch
from
+653 −25
Open

[fix][broker] Manage in-flight bucket work during clear #26401

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7f22e56
[fix][broker] Manage in-flight bucket work during clear
nodece Aug 24, 2026
e3c264a
[fix][broker] Clean up stale bucket cursor properties on recovery
nodece Aug 24, 2026
9781754
[fix][broker] Keep merge source snapshots when the merged creation fails
nodece Aug 24, 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
78 changes: 55 additions & 23 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 @@ -37,6 +37,7 @@
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
Expand Down Expand Up @@ -130,6 +131,8 @@ public static record SnapshotKey(long ledgerId, long entryId) {}

private CompletableFuture<Void> pendingLoad = null;

private final Set<CompletableFuture<Void>> pendingDeletes = ConcurrentHashMap.newKeySet();

private volatile CompletableFuture<Void> trimFuture;

public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Expand Down Expand Up @@ -255,8 +258,7 @@ private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryT
Range<Long> key = mapEntry.getKey();
ImmutableBucket immutableBucket = mapEntry.getValue();
removeBucket(key);
// delete asynchronously without waiting for completion
immutableBucket.asyncDeleteBucketSnapshot(stats);
trackDelete(immutableBucket.asyncDeleteBucketSnapshot(stats));
}

long totalLength = 0;
Expand Down Expand Up @@ -290,7 +292,9 @@ private CompletableFuture<List<DelayedIndex>> handleRecoverBucketSnapshotEntry(I
if (e == null) {
f.complete(v);
} else {
if (e instanceof BucketNotExistException) {
// Dependent stages wrap checked exceptions in CompletionException, so unwrap
// before matching the not-exist case.
if (FutureUtil.unwrapCompletionException(e) instanceof BucketNotExistException) {
// If the bucket does not exist, return an empty list,
// the bucket will be deleted from `immutableBuckets` in the next step.
f.complete(Collections.emptyList());
Expand Down Expand Up @@ -572,7 +576,7 @@ private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot(List<Immut
return CombinedSegmentDelayedIndexQueue.wrap(
getAllSnapshotFutures.stream().map(CompletableFuture::join).toList());
})
.thenAccept(combinedDelayedIndexQueue -> {
.thenCompose(combinedDelayedIndexQueue -> {
synchronized (BucketDelayedDeliveryTracker.this) {
long createStartTime = System.currentTimeMillis();
stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create);
Expand Down Expand Up @@ -604,17 +608,20 @@ private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot(List<Immut

afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime);

immutableBucketDelayedIndexPair.getLeft().getSnapshotCreateFuture()
.orElse(NULL_LONG_PROMISE).thenCompose(___ -> {
List<CompletableFuture<Void>> removeFutures =
buckets.stream().map(bucket -> bucket.asyncDeleteBucketSnapshot(stats))
.toList();
return FutureUtil.waitForAll(removeFutures);
});

for (ImmutableBucket bucket : buckets) {
removeBucket(Range.closed(bucket.getStartLedgerId(), bucket.getEndLedgerId()));
}

// A failed merged-bucket creation completes with INVALID_BUCKET_ID: keep
// the source snapshots so their delayed indexes stay recoverable.
return immutableBucketDelayedIndexPair.getLeft().getSnapshotCreateFuture()
.orElse(NULL_LONG_PROMISE)
.thenCompose(mergedBucketId -> INVALID_BUCKET_ID.equals(mergedBucketId)
? CompletableFuture.<Void>completedFuture(null)
: FutureUtil.waitForAll(
buckets.stream()
.map(bucket -> bucket.asyncDeleteBucketSnapshot(stats))
Comment on lines +617 to +623

@Denovo1998 Denovo1998 Aug 24, 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 source-deletion chain is gated by the post-processed snapshotCreateFuture; however, afterCreateImmutableBucket() converts a failed snapshot creation into a normally completed INVALID_BUCKET_ID. Consequently, this thenCompose still deletes all source snapshots even when the merged snapshot was never created. Because the source bucket mappings were previously removed, a restart following this path can lose the only durable delayed-index state.

There is also a second variant: ImmutableBucket.asyncSaveBucketSnapshot() currently suppresses putBucketKeyId() failure and still completes with the new bucket id, so a positive id does not guarantee the merged snapshot is recoverable.

Could source deletion be gated on successful durable handoff — including both snapshot persistence and cursor-property publication — while keeping the source buckets recoverable otherwise?

@nodece nodece Aug 24, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Agreed, the delete was wrong — fixed in 9781754: the source-delete chain now skips when the merged creation completed with INVALID_BUCKET_ID, so the source buckets stay recoverable from their cursor properties. Added a test that fails the merged creation and recreates the tracker with the same cursor/storage: both source buckets come back.

The second variant — putBucketKeyId() failing but still completing the create future with a positive id — is create-path semantics from master; changing it here felt too invasive. Even in that case delivery is not at risk: deliverAtTime is in the message metadata, and undelivered messages are just re-read from the backlog and re-tracked on restart (the same fallback the in-memory tracker uses). That part fits the follow-up / #26280.

.toList()));
}
});
});
Expand Down Expand Up @@ -715,13 +722,12 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
long loadStartTime = System.currentTimeMillis();
stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.load);
CompletableFuture<Void> loadFuture = pendingLoad = bucket.asyncLoadNextBucketSnapshotEntry()
.thenAccept(indexList -> {
.thenCompose(indexList -> {
synchronized (BucketDelayedDeliveryTracker.this) {
this.snapshotSegmentLastIndexMap.remove(snapshotKey);
if (CollectionUtils.isEmpty(indexList)) {
removeBucket(Range.closed(bucket.getStartLedgerId(), bucket.getEndLedgerId()));
bucket.asyncDeleteBucketSnapshot(stats);
return;
return bucket.asyncDeleteBucketSnapshot(stats);
}
DelayedIndex
lastDelayedIndex = indexList.get(indexList.size() - 1);
Expand All @@ -732,6 +738,7 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
sharedBucketPriorityQueue.add(index.getTimestamp(), index.getLedgerId(),
index.getEntryId());
}
return CompletableFuture.completedFuture(null);
}
}).whenComplete((__, ex) -> {
if (ex != null) {
Expand Down Expand Up @@ -786,15 +793,25 @@ private synchronized boolean checkPendingLoadDone() {
return false;
}

private void trackDelete(CompletableFuture<Void> deleteFuture) {
synchronized (this) {
pendingDeletes.add(deleteFuture);
}
deleteFuture.whenComplete((__, ex) -> {
synchronized (BucketDelayedDeliveryTracker.this) {
pendingDeletes.remove(deleteFuture);
}
});
}

@Override
public boolean shouldPauseAllDeliveries() {
return false;
}

@Override
public synchronized CompletableFuture<Void> clear() {
// Wait for any in-flight trim+merge to settle, then clear.
// Reuse trimFuture to block new triggers until the clear chain completes.
// Wait for in-flight trim/merge and pending load/delete work before resetting state.
CompletableFuture<Void> before = trimFuture != null && !trimFuture.isDone()
? trimFuture : CompletableFuture.completedFuture(null);
trimFuture = before
Expand All @@ -803,14 +820,29 @@ public synchronized CompletableFuture<Void> clear() {
return null;
})
.thenCompose(__ -> {
List<CompletableFuture<?>> pending = new ArrayList<>();

@Denovo1998 Denovo1998 Aug 23, 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 takes a one-time snapshot of pendingLoad and pendingDeletes, but clear() does not mark the tracker as clearing or prevent getScheduledMessages() from starting another load.

While clear() waits for trimFuture or a captured load, the dispatcher or timer may still enter the tracker. Once the captured load completes, another call could acquire the tracker lock, see that pendingLoad is finished, and schedule the next load before the cleanup routine runs. That new callback could then repopulate queued and mapped items after lines 835–838 clear them. shouldPauseAllDeliveries() would also remain false.

Could clear() establish a lifecycle generation or clearing state synchronously on entry, prevent new loads during clearing, and make every async callback verify the generation before publishing its state?

@nodece nodece Aug 24, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thanks. This PR intentionally fixes the quiescence snapshot: clear() now waits for trimFuture, the current pendingLoad, and tracked pendingDeletes before resetting state, with tests for segment load, terminal-load delete, merge create/load/delete, and failed tracked deletes. A full clearing lifecycle/generation fence is broader than inflight-future management and is planned for the follow-up PR; it will prevent newly scheduled loads from starting during clear and verify generation before publishing callbacks.

@Denovo1998 Denovo1998 Aug 24, 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.

I re‑checked the production call path. PersistentSubscription.clearBacklog() calls dispatcher.clearDelayedMessages() while the dispatcher remains usable, and the normal dispatcher path can still call getScheduledMessages().

Thus, a single snapshot of pendingLoad / pendingDeletes does not serve as a quiescence barrier: after the captured load completes, another load can begin before the cleanup continuation acquires the tracker lock, and that callback could publish state after clear() has finished.

This appears to fall directly within the guarantee stated in this PR, not just as a broader reset/ownership issue. Could clear() establish a clearing generation or state before releasing the lock, prevent new async work from starting, and require pre‑clear callbacks to verify that generation before publishing?

@nodece nodece Aug 24, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I'd still keep this one out of this PR. What this change promises is that clear() drains everything that was in flight when it entered. A load scheduled after that is new-lifecycle work, and what it publishes after the reset is inert — entries get popped with removeIndexBit() returning false and silently dropped (never delivered), and the boundary entries point at buckets that are no longer in immutableBuckets, so they can't chain more loads. It burns a few wasted reads and stops by itself. Master today doesn't drain anything at all, so this is strictly narrower already.

Doing what you suggest properly — generation + blocking new work, verified at publish — is a real lifecycle change of its own, and I'd rather not bolt it onto this one. I'll take it in the follow-up, happy to discuss the shape there.

synchronized (BucketDelayedDeliveryTracker.this) {
CompletableFuture<Void> future = cleanImmutableBuckets();
sharedBucketPriorityQueue.clear();
index.clear();
lastMutableBucket.clear();
snapshotSegmentLastIndexMap.clear();
return future;
if (pendingLoad != null) {
pending.add(pendingLoad);
}
pending.addAll(pendingDeletes);
}
return FutureUtil.waitForAll(pending)
.exceptionally(t -> {
log.warn().exception(t)
.log("Failed to wait for pending delayed delivery work, but still clear");
return null;
})
Comment on lines +830 to +835

@Denovo1998 Denovo1998 Aug 24, 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.

I think clear() may still return success even when some snapshot cleanup actually failed.

asyncDeleteBucketSnapshot() only removes the cursor property after the snapshot delete succeeds. But in both the terminal-load path and merge path, the bucket is already removed from immutableBuckets before the delete finishes.

So if the delete keeps failing after retries, clear() ignores the error here and continues. At that point cleanImmutableBuckets() also can't find this bucket anymore. The old cursor property may still be there, and after recreating the tracker the old bucket could be recovered again.

The current tests only check that clear() completes successfully. Could we also recreate the tracker with the same cursor/storage and verify that the deleted bucket isn't recovered again?

I think either clear() should propagate this cleanup failure, or we need to keep track of the failed cleanup until its cursor property is actually removed.

@nodece nodece Aug 24, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Looked at both failure shapes, I think they both converge:

  • delete keeps failing: snapshot and property survive as a consistent pair — recovery just loads the bucket again (it still exists) and the next trim/clear retries the cleanup.
  • delete ok but property removal failed: recovery resolves the property, hits BucketNotExist, and the toBeDeleted path in recoverBucketSnapshot() removes it.

While writing the recreate test for the second shape I found that path was actually dead: the chain delivers the BucketNotExistException wrapped in a CompletionException, so the instanceof never matched and recovery would just throw. Fixed by unwrapping before the check (e3c264a), and the test now recreates the tracker with the same cursor/storage and watches the stale property get cleaned up — plus a test for the consistent-pair shape, where the bucket just loads normally again.

.thenCompose(ignore -> {
synchronized (BucketDelayedDeliveryTracker.this) {
CompletableFuture<Void> future = cleanImmutableBuckets();
sharedBucketPriorityQueue.clear();
index.clear();
lastMutableBucket.clear();
snapshotSegmentLastIndexMap.clear();
return future;
}
});
});
return trimFuture;
}
Expand Down
4 changes: 3 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,9 @@ public CompletableFuture<List<SnapshotSegment>> getBucketSnapshotSegment(long bu
for (int i = (int) firstSegmentEntryId; i <= lastEntryId; i++) {
ByteBuf byteBuf = this.bucketSnapshots.get(bucketId).get(i);
SnapshotSegment snapshotSegment = new SnapshotSegment();
snapshotSegment.parseFrom(byteBuf, byteBuf.readableBytes());
// Parse from a slice so the stored buffer stays re-readable, like a real storage.
ByteBuf slice = byteBuf.slice();
snapshotSegment.parseFrom(slice, slice.readableBytes());
snapshotSegments.add(snapshotSegment);
}
return snapshotSegments;
Expand Down
Loading
Loading