You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "nodece (via GitHub)" <gi...@apache.org> on 2026/07/27 07:48:43 UTC
[PR] [fix][broker] Fix delayed delivery bucket merge skipping snapshot segments [pulsar]
nodece opened a new pull request, #26251:
URL: https://github.com/apache/pulsar/pull/26251
### Motivation
During bucket delayed delivery merge, old bucket snapshots are loaded, combined into a merged bucket, and then deleted from BookKeeper.
`ImmutableBucket.getRemainSnapshotSegment()` loaded snapshot segments starting from `currentSegmentEntryId + 1`, which skipped segments that had already been loaded into memory. Since the old bucket snapshots were removed after merge, the skipped segments were not included in the merged bucket snapshot.
After tracker recovery, the merged bucket snapshot could not recover the missing segment data. Delayed messages rely on the delayed delivery tracker for scheduling.
When a segment is missing from the merged bucket snapshot, the corresponding positions are not restored into the delayed delivery tracker. Messages that have already expired can bypass delayed delivery tracking and be dispatched normally, but messages that have not expired may never be scheduled because their positions are no longer tracked.
### Modifications
* Rename `ImmutableBucket.getRemainSnapshotSegment()` to `getAllSnapshotSegments()` and load snapshot segments starting from segment 1 instead of `currentSegmentEntryId + 1`, ensuring that all segment data is included in the merged bucket snapshot.
* Add `testMergePreservesAllSnapshotSegments` to verify that all snapshot segments are preserved after merge and recovery.
* Increase Awaitility timeout to 30s in `testMergeSnapshot` and `testWithBkException`, as loading all segments requires additional lazy-load cycles.
--
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] [fix][broker] Fix bucket delayed delivery state consistency issues [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#issuecomment-5113564362
Thanks for the update. The main correctness issues are addressed. LGTM.
--
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] [fix][broker] Fix bucket delayed delivery counter and dedup consistency [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#discussion_r3719921866
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -682,9 +679,7 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
long entryId = sharedBucketPriorityQueue.peekN3();
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
sharedBucketPriorityQueue.pop();
- if (removeIndexBit(ledgerId, entryId)) {
- numberDelayedMessages.decrementAndGet();
- }
+ removeIndexBit(ledgerId, entryId);
continue;
}
Review Comment:
@nodece Okay. You can ping me after creating the new PR.
--
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] [fix][broker] Fix bucket delayed delivery counter and dedup consistency [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#discussion_r3725615841
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -682,9 +679,7 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
long entryId = sharedBucketPriorityQueue.peekN3();
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
sharedBucketPriorityQueue.pop();
- if (removeIndexBit(ledgerId, entryId)) {
- numberDelayedMessages.decrementAndGet();
- }
+ removeIndexBit(ledgerId, entryId);
continue;
}
Review Comment:
Link: https://github.com/apache/pulsar/pull/26280
--
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] [fix][broker] Fix bucket delayed delivery counter and dedup consistency [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #26251:
URL: https://github.com/apache/pulsar/pull/26251
--
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] [fix][broker] Fix bucket delayed delivery state consistency issues [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#discussion_r3704433118
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -682,9 +679,7 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
long entryId = sharedBucketPriorityQueue.peekN3();
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
sharedBucketPriorityQueue.pop();
- if (removeIndexBit(ledgerId, entryId)) {
- numberDelayedMessages.decrementAndGet();
- }
+ removeIndexBit(ledgerId, entryId);
continue;
}
Review Comment:
This branch runs before the snapshot-segment boundary handling below. If the last index of the currently loaded segment belongs to a ledger below firstLiveLedgerId, we pop it and continue, so asyncLoadNextBucketSnapshotEntry() is never invoked.
This is especially problematic for a bucket retained because it overlaps the first live ledger. For example:
segment 1 contains only indexes from ledger 1;
firstLiveLedgerId is 3;
segment 2 contains valid delayed indexes from ledger 3 or later.
After the segment-1 tail is filtered, the boundary mapping still points to the removed queue key, while segment 2 remains only in snapshot storage. The in-memory queues can become empty and the valid delayed messages may never be loaded. The Long.MAX_VALUE guard from #26080 prevents a crash but does not restore delivery progress.
Could we process the snapshot-segment boundary before skipping an orphaned entry, while still suppressing delivery of that entry? A regression test should use a multi-segment bucket whose first segment is fully below firstLiveLedgerId and whose next segment contains live-ledger messages.
--
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] [fix][broker] Fix bucket delayed delivery state consistency issues [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#issuecomment-5112232116
@void-ptr974 This PR body has been updated.
--
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] [fix][broker] Fix bucket delayed delivery counter and dedup consistency [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#discussion_r3710168248
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -682,9 +679,7 @@ public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages)
long entryId = sharedBucketPriorityQueue.peekN3();
if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
sharedBucketPriorityQueue.pop();
- if (removeIndexBit(ledgerId, entryId)) {
- numberDelayedMessages.decrementAndGet();
- }
+ removeIndexBit(ledgerId, entryId);
continue;
}
Review Comment:
@Denovo1998 Good catch, I will open a new PR to handle this issue.
--
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] [fix][broker] Fix delayed delivery bucket merge skipping snapshot segments [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#issuecomment-5090252482
Link: https://github.com/apache/pulsar/issues/19902
--
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] [fix][broker] Fix delayed delivery bucket merge skipping snapshot segments [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26251:
URL: https://github.com/apache/pulsar/pull/26251#issuecomment-5092287473
When a selected bucket still has its current segment loaded in `sharedBucketPriorityQueue`, reading from segment 1 reads that same segment from storage again. The merge then creates a new bucket; creating a bucket puts the first segment of its new snapshot into `sharedBucketPriorityQueue`, but the entries already queued for the source buckets are not removed.
For example, assume a source bucket's current segment contains `1:1` with a delivery time of `10ms`. Before the merge, one copy of `1:1` is already in the shared queue. The merged snapshot also contains `1:1`, and its first segment enqueues another copy. At `10ms`, `getScheduledMessages(1)` can return `1:1`; the next call can return `1:1` again. The second dequeue also decrements `numberDelayedMessages` again, so the count becomes incorrect.
If `currentSegmentEntryId` has advanced, reading all segments also includes older segments that have already been dispatched. Those entries can be persisted in the merged snapshot and replayed after a restart.
We need to keep the current segment recoverable without adding a second live queue entry, and exclude entries that have already been dispatched.
--
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