You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "dao-jun (via GitHub)" <gi...@apache.org> on 2026/06/09 19:18:50 UTC
[PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
dao-jun opened a new pull request, #25984:
URL: https://github.com/apache/pulsar/pull/25984
### Motivation
When a topic's managed ledger trims old ledgers (via retention or compaction), the immutable bucket snapshots in BucketDelayedDeliveryTracker that cover those deleted ledger ranges become orphaned — their underlying message data no longer exists, yet they:
1. Continue to occupy storage as bucket snapshots.
2. Count against the maxNumBuckets limit, potentially triggering unnecessary and expensive merge operations.
Prior to this change, the tracker only merged buckets when the count exceeded the limit; it never cleaned up buckets whose ledger range had been fully trimmed. This change introduces a trim step that runs before merge, deleting orphaned buckets and decrementing the delayed-message counter accordingly.
### Modifications
Source (BucketDelayedDeliveryTracker.java)
- Added trimFuture: A volatile CompletableFuture<Void> that represents the currently in-flight trim/merge/clear operation. It acts as both a gate (preventing concurrent trim chains) and a synchronization point for clear().
- Added asyncTrimImmutableBuckets(): Scans the managed ledger's surviving ledger range. Buckets whose upperEndpoint falls entirely before the earliest surviving ledger are selected for deletion — these are orphaned and safe to remove.
- Added deleteBucketSnapshot(): Removes a single bucket from immutableBuckets, asynchronously deletes its snapshot, and re-adds the bucket on failure (rolling back the message count). Failures propagate via CompletionException to stop the sequential deletion chain, avoiding wasted attempts when storage is unavailable.
- Changed the merge trigger in addMessage(): Before merging, the tracker now runs asyncTrimImmutableBuckets() first. The trimFuture gate ensures only one trim/merge chain is in-flight at a time.
- Added early-return to asyncMergeBucketSnapshot(): If the bucket count is already within the limit, merge returns immediately without scanning for merge candidates.
- Rewrote clear(): Instead of immediately clearing state (which could race with an in-flight trim's failure callback), clear() now chains itself after any pending trimFuture. It also sets trimFuture to the clear chain, blocking new trim triggers until the clear completes.
Test (BucketDelayedDeliveryTrackerTest.java)
Added 4 tests using a helper that injects a mocked ManagedLedger:
| Test | Scenario |
| -- | -- |
| testTrimRemovesOrphanedBuckets | Buckets covering ledgers below firstLedgerId are removed, remaining ones have ranges ≥ firstLedgerId |
| testTrimHandlesDeleteFailure | Injected delete exception is consumed, tracker remains operational |
| testTrimWithNoOrphanedBuckets | With firstLedgerId=0, trim is a no-op, merge runs normally |
| testMergeEarlyReturnWhenWithinLimit | With maxNumBuckets far above actual count, merge returns early without compaction |
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#issuecomment-4697487022
I think there is still an issue here: `getScheduledMessages()` only skips entries before `firstLiveLedgerId` when `!containsMessage(...)`.
For old-ledger entries that are still present in the bucket bitmap, `containsMessage(...)` returns true, so those deleted-ledger positions can still be returned. The key point is that `ledgerId < firstLiveLedgerId` should be enough to skip the entry, and the tracker state/counter should stay consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3457012814
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -634,13 +650,16 @@ 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 && !containsMessage(ledgerId, entryId)) {
+ sharedBucketPriorityQueue.pop();
Review Comment:
```suggestion
sharedBucketPriorityQueue.pop();
removeIndexBit(ledgerId, entryId)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3456994539
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -634,13 +650,16 @@ 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 && !containsMessage(ledgerId, entryId)) {
Review Comment:
```suggestion
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
```
`!containsMessage(ledgerId, entryId)` looks unecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3456986575
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -816,4 +849,60 @@ public Map<String, TopicMetricBean> genTopicMetricMap() {
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 = context.getCursor().getManagedLedger();
+
+ Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets = immutableBuckets.asMapOfRanges().entrySet().stream()
+ .filter(e -> e.getKey().upperEndpoint() < firstLedgerId)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ 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().attr("LedgerName", ledgerName)
+ .attr("BucketKey", bucket.bucketKey())
+ .log("Failed to delete bucket snapshot");
+ 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() {
Review Comment:
Would it make sense to return `context.getCursor().getMarkDeletedPosition()` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3397601506
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -754,12 +769,21 @@ public boolean shouldPauseAllDeliveries() {
@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.thenCompose(__ -> {
Review Comment:
Good catch! That’s a bug for sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3457012814
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -634,13 +650,16 @@ 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 && !containsMessage(ledgerId, entryId)) {
+ sharedBucketPriorityQueue.pop();
Review Comment:
```suggestion
sharedBucketPriorityQueue.pop();
if removeIndexBit(ledgerId, entryId) {
numberDelayedMessages.decrementAndGet();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3396557389
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -816,4 +840,58 @@ public Map<String, TopicMetricBean> genTopicMetricMap() {
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() {
+ ManagedLedger ledger = context.getCursor().getManagedLedger();
+ if (ledger == null || ledger.getLedgersInfo().isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ Long firstLedgerId = ledger.getLedgersInfo().firstKey();
+ if (null == firstLedgerId) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets = immutableBuckets.asMapOfRanges().entrySet().stream()
+ .filter(e -> e.getKey().upperEndpoint() < firstLedgerId)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ 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) {
+ synchronized (this) {
+ immutableBuckets.remove(range);
+ numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
+ }
Review Comment:
Removing a bucket from immutableBuckets might not be enough. When a mutable bucket is sealed, the first snapshot segment has already been loaded into sharedBucketPriorityQueue, and snapshotSegmentLastIndexMap can still point to this bucket. After this trim removes the range and decrements numberDelayedMessages, getScheduledMessages() can still pop those queued entries and return positions from ledgers that have already been deleted, then decrement numberDelayedMessages again.
Could we either purge the in-memory state for the trimmed bucket under the same lock, such as rebuilding sharedBucketPriorityQueue and removing matching snapshotSegmentLastIndexMap entries, or otherwise avoid trimming buckets that still have loaded segments? Please also add a test that advances the clock after trim and asserts no positions below firstLedgerId are returned and the delayed-message counter remains consistent.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java:
##########
@@ -469,4 +471,137 @@ public void testClear(BucketDelayedDeliveryTracker tracker)
tracker.close();
}
+
+ private static class TrackerWithStorage {
+ final BucketDelayedDeliveryTracker tracker;
+ final MockBucketSnapshotStorage storage;
+
+ TrackerWithStorage(BucketDelayedDeliveryTracker tracker, MockBucketSnapshotStorage storage) {
+ this.tracker = tracker;
+ this.storage = storage;
+ }
+
+ void close() throws Exception {
+ tracker.close();
+ storage.close();
+ }
+ }
+
+ private TrackerWithStorage createTrackerWithMockLedger(long firstLedgerId, int maxNumBuckets)
+ throws Exception {
+ MockBucketSnapshotStorage storage = new MockBucketSnapshotStorage();
+ 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;
+ }
+ };
+
+ 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);
+ }
+
+ @Test
+ public void testTrimRemovesOrphanedBuckets() throws Exception {
+ TrackerWithStorage ts = createTrackerWithMockLedger(50L, 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 after trim+merge");
+
+ ts.tracker.getImmutableBuckets().asMapOfRanges().forEach((range, bucket) ->
+ assertTrue(range.lowerEndpoint() >= 50L,
+ "Remaining bucket range " + range + " should be >= 50"));
+
Review Comment:
This test only checks that immutableBuckets no longer contains ranges below firstLedgerId. It does not verify the externally visible behavior after trim. Please advance the mock clock and call getScheduledMessages(), then assert that no position from ledgers below firstLedgerId is returned and that getNumberOfDelayedMessages() remains consistent. That would catch stale entries left in sharedBucketPriorityQueue after the bucket range is removed.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -754,12 +769,21 @@ public boolean shouldPauseAllDeliveries() {
@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.thenCompose(__ -> {
Review Comment:
If clear() is called while trimFuture is still in flight, and that trim/merge chain later completes exceptionally, this before.thenCompose(...) will not execute the clear block. The whenComplete in addMessage only logs the failure and preserves the exceptional completion, so clear() can return exceptionally while leaving immutableBuckets, sharedBucketPriorityQueue, lastMutableBucket, snapshotSegmentLastIndexMap, and numberDelayedMessages uncleared.
Could we normalize the previous future before chaining clear, for example with handle/exceptionally, so clear always runs after the in-flight trim/merge settles?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3398448012
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -816,4 +840,58 @@ public Map<String, TopicMetricBean> genTopicMetricMap() {
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() {
+ ManagedLedger ledger = context.getCursor().getManagedLedger();
+ if (ledger == null || ledger.getLedgersInfo().isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ Long firstLedgerId = ledger.getLedgersInfo().firstKey();
+ if (null == firstLedgerId) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets = immutableBuckets.asMapOfRanges().entrySet().stream()
+ .filter(e -> e.getKey().upperEndpoint() < firstLedgerId)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ 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) {
+ synchronized (this) {
+ immutableBuckets.remove(range);
+ numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
+ }
Review Comment:
Not clearing the data of `sharedBucketPriorityQueue` is intentionally designed, as `ManagedCursorImpl.asyncReplayEntries` will filter out invalid Positions.
I did miss cleaning the `snapshotSegmentLastIndexMap`, this is a problem.
Double decrease `numberDelayedMessages` is indeed a problem, good catch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #25984:
URL: https://github.com/apache/pulsar/pull/25984
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#issuecomment-4779933370
#26080 is also required when cherry-picking
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Trim orphaned bucket snapshots when ledgers are deleted [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #25984:
URL: https://github.com/apache/pulsar/pull/25984#discussion_r3456991127
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -816,4 +849,60 @@ public Map<String, TopicMetricBean> genTopicMetricMap() {
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 = context.getCursor().getManagedLedger();
+
+ Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets = immutableBuckets.asMapOfRanges().entrySet().stream()
+ .filter(e -> e.getKey().upperEndpoint() < firstLedgerId)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Review Comment:
```suggestion
Map<Range<Long>, ImmutableBucket> toBeDeletedBuckets = new HashMap<>(immutableBuckets.subRangeMap(Range.lessThan(firstLedgerId)).asMapOfRanges());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org