gaoran10 commented on code in PR #19138:
URL: https://github.com/apache/pulsar/pull/19138#discussion_r1066725325
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java:
##########
@@ -243,6 +263,53 @@ public synchronized boolean addMessage(long ledgerId, long
entryId, long deliver
return true;
}
+ private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot() {
+ List<ImmutableBucket> values =
immutableBuckets.asMapOfRanges().values().stream().toList();
+ long minNumberMessages = Long.MAX_VALUE;
+ int minIndex = -1;
+ for (int i = 0; i + 1 < values.size(); i++) {
+ ImmutableBucket bucketL = values.get(i);
+ ImmutableBucket bucketR = values.get(i + 1);
+ long numberMessages = bucketL.numberBucketDelayedMessages +
bucketR.numberBucketDelayedMessages;
+ if (numberMessages < minNumberMessages) {
+ minNumberMessages = (int) numberMessages;
+ minIndex = i;
+ }
+ }
+ return asyncMergeBucketSnapshot(values.get(minIndex),
values.get(minIndex + 1));
+ }
+
+ private synchronized CompletableFuture<Void>
asyncMergeBucketSnapshot(ImmutableBucket bucketA,
+
ImmutableBucket bucketB) {
+ immutableBuckets.remove(Range.closed(bucketA.startLedgerId,
bucketA.endLedgerId));
+ immutableBuckets.remove(Range.closed(bucketB.startLedgerId,
bucketB.endLedgerId));
+
+ CompletableFuture<Long> snapshotCreateFutureA =
+
bucketA.getSnapshotCreateFuture().orElse(CompletableFuture.completedFuture(null));
+ CompletableFuture<Long> snapshotCreateFutureB =
+
bucketB.getSnapshotCreateFuture().orElse(CompletableFuture.completedFuture(null));
+
+ return CompletableFuture.allOf(snapshotCreateFutureA,
snapshotCreateFutureB).thenCompose(__ -> {
+
CompletableFuture<List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment>>
futureA =
+ bucketA.getRemainSnapshotSegment();
+
CompletableFuture<List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment>>
futureB =
+ bucketB.getRemainSnapshotSegment();
+ return futureA.thenCombine(futureB,
CombinedSegmentDelayedIndexQueue::wrap)
+ .thenCompose(combinedDelayedIndexQueue -> {
+ CompletableFuture<Void> removeAFuture =
bucketA.asyncDeleteBucketSnapshot();
+ CompletableFuture<Void> removeBFuture =
bucketB.asyncDeleteBucketSnapshot();
+
+ return CompletableFuture.allOf(removeAFuture,
removeBFuture).thenRun(() -> {
Review Comment:
1. Delete the original bucket index first, then save the newly merged index,
this may lose the bucket index, but the tracker will rebuild the index.
2. Save the newly merged index first, then delete the original bucket index,
this may cause intersecting buckets, and we need to find a way to clean up the
original buckets.
OK, I think there are two ways, the first way is simple and clear.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
