From 9225790cce0cb00d844c6831fa5aa7255e0f34ad Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Mon, 20 Jul 2026 18:26:28 +0800 Subject: [PATCH 1/3] [improve][broker] Refactor bucket delayed-delivery class structure --- .../pulsar/broker/delayed/bucket/Bucket.java | 182 ------------------ .../broker/delayed/bucket/BucketContext.java | 29 +++ .../bucket/BucketDelayedDeliveryTracker.java | 113 +++++++---- .../delayed/bucket/ImmutableBucket.java | 176 +++++++++++++++-- .../broker/delayed/bucket/MutableBucket.java | 32 ++- 5 files changed, 275 insertions(+), 257 deletions(-) delete mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketContext.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java deleted file mode 100644 index 452e6b13554a6..0000000000000 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java +++ /dev/null @@ -1,182 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.broker.delayed.bucket; - -import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry; -import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.DELAYED_BUCKET_KEY_PREFIX; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import lombok.AllArgsConstructor; -import lombok.CustomLog; -import lombok.Data; -import org.apache.bookkeeper.mledger.ManagedCursor; -import org.apache.bookkeeper.mledger.ManagedLedgerException; -import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata; -import org.apache.pulsar.broker.delayed.proto.SnapshotSegment; -import org.apache.pulsar.common.util.Codec; -import org.apache.pulsar.common.util.FutureUtil; -import org.apache.pulsar.common.util.collections.LongBitmap; -import org.apache.pulsar.common.util.collections.LongBitmaps; - -@CustomLog -@Data -@AllArgsConstructor -abstract class Bucket { - - static final String DELIMITER = "_"; - static final int MaxRetryTimes = 3; - - protected final String dispatcherName; - - protected final ManagedCursor cursor; - - protected final FutureUtil.Sequencer sequencer; - - protected final BucketSnapshotStorage bucketSnapshotStorage; - - long startLedgerId; - long endLedgerId; - - Map delayedIndexBitMap; - - long numberBucketDelayedMessages; - - int lastSegmentEntryId; - - volatile int currentSegmentEntryId; - - volatile long snapshotLength; - - private volatile Long bucketId; - - private volatile CompletableFuture snapshotCreateFuture; - - Bucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, - BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { - this(dispatcherName, cursor, sequencer, storage, startLedgerId, endLedgerId, new HashMap<>(), -1, -1, 0, 0, - null, null); - } - - boolean containsMessage(long ledgerId, long entryId) { - LongBitmap bitSet = delayedIndexBitMap.get(ledgerId); - if (bitSet == null) { - return false; - } - return bitSet.contains(entryId); - } - - void putIndexBit(long ledgerId, long entryId) { - delayedIndexBitMap.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId); - } - - boolean removeIndexBit(long ledgerId, long entryId) { - LongBitmap bitSet = delayedIndexBitMap.get(ledgerId); - if (bitSet == null) { - return false; - } - boolean contained = bitSet.remove(entryId); - if (contained) { - if (bitSet.isEmpty()) { - delayedIndexBitMap.remove(ledgerId); - } - - if (numberBucketDelayedMessages > 0) { - numberBucketDelayedMessages--; - } - } - return contained; - } - - String bucketKey() { - return String.join(DELIMITER, DELAYED_BUCKET_KEY_PREFIX, String.valueOf(startLedgerId), - String.valueOf(endLedgerId)); - } - - Optional> getSnapshotCreateFuture() { - return Optional.ofNullable(snapshotCreateFuture); - } - - Optional getBucketId() { - return Optional.ofNullable(bucketId); - } - - long getAndUpdateBucketId() { - Optional bucketIdOptional = getBucketId(); - if (bucketIdOptional.isPresent()) { - return bucketIdOptional.get(); - } - - String bucketIdStr = cursor.getCursorProperties().get(bucketKey()); - long bucketId = Long.parseLong(bucketIdStr); - setBucketId(bucketId); - return bucketId; - } - - CompletableFuture asyncSaveBucketSnapshot( - ImmutableBucket bucket, SnapshotMetadata snapshotMetadata, - List bucketSnapshotSegments) { - final String bucketKey = bucket.bucketKey(); - final String cursorName = Codec.decode(cursor.getName()); - final String topicName = dispatcherName.substring(0, dispatcherName.lastIndexOf(" / " + cursorName)); - return executeWithRetry( - () -> bucketSnapshotStorage.createBucketSnapshot(snapshotMetadata, bucketSnapshotSegments, bucketKey, - topicName, cursorName) - .whenComplete((__, ex) -> { - if (ex != null) { - log.warn() - .attr("dispatcher", dispatcherName) - .attr("bucketKey", bucketKey) - .exception(ex) - .log("Failed to create bucket snapshot"); - } - }), BucketSnapshotPersistenceException.class, MaxRetryTimes).thenCompose(newBucketId -> { - bucket.setBucketId(newBucketId); - - return putBucketKeyId(bucketKey, newBucketId).exceptionally(ex -> { - log.warn() - .attr("dispatcher", dispatcherName) - .attr("bucketKey", bucketKey) - .attr("bucketId", newBucketId) - .exception(ex) - .log("Failed to record bucketId to cursor property"); - return null; - }).thenApply(__ -> newBucketId); - }); - } - - private CompletableFuture putBucketKeyId(String bucketKey, Long bucketId) { - if (bucketId == null) { - return FutureUtil.failedFuture(new NullPointerException("Expected bucketId should not be null")); - } - return sequencer.sequential(() -> { - return executeWithRetry(() -> cursor.putCursorProperty(bucketKey, String.valueOf(bucketId)), - ManagedLedgerException.BadVersionException.class, MaxRetryTimes); - }); - } - - protected CompletableFuture removeBucketCursorProperty(String bucketKey) { - return sequencer.sequential(() -> { - return executeWithRetry(() -> cursor.removeCursorProperty(bucketKey), - ManagedLedgerException.BadVersionException.class, MaxRetryTimes); - }); - } -} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketContext.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketContext.java new file mode 100644 index 0000000000000..cf1d7bfe981cb --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketContext.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; + +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.pulsar.common.util.FutureUtil; + +record BucketContext( + String dispatcherName, + ManagedCursor cursor, + FutureUtil.Sequencer sequencer, + BucketSnapshotStorage bucketSnapshotStorage) { +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java index 400e60e24a184..a8b1fd3dc9b4a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java @@ -20,7 +20,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static org.apache.bookkeeper.mledger.ManagedCursor.CURSOR_INTERNAL_PROPERTY_PREFIX; -import static org.apache.pulsar.broker.delayed.bucket.Bucket.DELIMITER; +import static org.apache.pulsar.broker.delayed.bucket.ImmutableBucket.DELIMITER; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Range; import com.google.common.collect.RangeMap; @@ -28,6 +28,8 @@ import io.github.merlimat.slog.Logger; import io.netty.util.Timeout; import io.netty.util.Timer; +import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; import java.time.Clock; import java.util.ArrayList; import java.util.Collections; @@ -35,7 +37,6 @@ import java.util.List; import java.util.Map; import java.util.NavigableSet; -import java.util.Optional; import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -64,6 +65,7 @@ import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.LongBitmap; +import org.apache.pulsar.common.util.collections.LongBitmaps; import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; @ThreadSafe @@ -98,6 +100,10 @@ public static record SnapshotKey(long ledgerId, long entryId) {} private final AtomicLong numberDelayedMessages = new AtomicLong(0); + @Getter + @VisibleForTesting + private final BucketContext ctx; + @Getter @VisibleForTesting private final MutableBucket lastMutableBucket; @@ -110,6 +116,10 @@ public static record SnapshotKey(long ledgerId, long entryId) {} @VisibleForTesting private final RangeMap immutableBuckets; + @Getter + @VisibleForTesting + private final Long2ObjectMap inflightIndex; + @Getter @VisibleForTesting private final AtomicLong bucketsCount = new AtomicLong(0); @@ -166,10 +176,11 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, this.maxNumBuckets = maxNumBuckets; this.sharedBucketPriorityQueue = new TripleLongPriorityQueue(); this.immutableBuckets = TreeRangeMap.create(); + this.inflightIndex = new Long2ObjectOpenHashMap<>(); this.snapshotSegmentLastIndexMap = new ConcurrentHashMap<>(); - this.lastMutableBucket = - new MutableBucket(context.getName(), context.getCursor(), FutureUtil.Sequencer.create(), - bucketSnapshotStorage); + this.ctx = new BucketContext(context.getName(), context.getCursor(), FutureUtil.Sequencer.create(), + bucketSnapshotStorage); + this.lastMutableBucket = new MutableBucket(ctx); this.stats = new BucketDelayedMessageIndexStats(); // Close the tracker if failed to recover. @@ -183,24 +194,21 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, } private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryTrackerException { - ManagedCursor cursor = this.lastMutableBucket.getCursor(); + ManagedCursor cursor = ctx.cursor(); Map cursorProperties = cursor.getCursorProperties(); if (MapUtils.isEmpty(cursorProperties)) { log.info("Recover delayed message index bucket snapshot finish, don't find bucket snapshot"); return 0; } - FutureUtil.Sequencer sequencer = this.lastMutableBucket.getSequencer(); Map, ImmutableBucket> toBeDeletedBucketMap = new HashMap<>(); cursorProperties.keySet().forEach(key -> { if (key.startsWith(DELAYED_BUCKET_KEY_PREFIX)) { String[] keys = key.split(DELIMITER); checkArgument(keys.length == 3); ImmutableBucket immutableBucket = - new ImmutableBucket(context.getName(), cursor, sequencer, - this.lastMutableBucket.bucketSnapshotStorage, - Long.parseLong(keys[1]), Long.parseLong(keys[2])); - putAndCleanOverlapRange(Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId), - immutableBucket, toBeDeletedBucketMap); + new ImmutableBucket(ctx, Long.parseLong(keys[1]), Long.parseLong(keys[2])); + putAndCleanOverlapRange(Range.closed(immutableBucket.getStartLedgerId(), + immutableBucket.getEndLedgerId()), immutableBucket, toBeDeletedBucketMap); } }); @@ -260,7 +268,7 @@ private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryT MutableLong numberDelayedMessages = new MutableLong(0); long totalLength = 0; for (ImmutableBucket bucket : immutableBucketMap.values()) { - numberDelayedMessages.add(bucket.numberBucketDelayedMessages); + numberDelayedMessages.add(bucket.getNumberBucketDelayedMessages()); totalLength += bucket.getSnapshotLength(); } totalSnapshotLengthBytes.set(totalLength); @@ -309,7 +317,7 @@ private synchronized void putAndCleanOverlapRange(Range range, ImmutableBu for (Map.Entry, ImmutableBucket> rangeEntry : subRangeMap.entrySet()) { // Use original key instead of truncated key for encloses check ImmutableBucket bucket = rangeEntry.getValue(); - Range originalKey = Range.closed(bucket.startLedgerId, bucket.endLedgerId); + Range originalKey = Range.closed(bucket.getStartLedgerId(), bucket.getEndLedgerId()); if (range.encloses(originalKey)) { toBeDeletedBucketMap.put(originalKey, bucket); @@ -336,21 +344,41 @@ public void run(Timeout timeout) throws Exception { super.run(timeout); } - private Optional findImmutableBucket(long ledgerId) { - if (immutableBuckets.asMapOfRanges().isEmpty()) { - return Optional.empty(); + private ImmutableBucket findImmutableBucket(long ledgerId) { + return immutableBuckets.get(ledgerId); + } + + private void trackInflight(long ledgerId, long entryId) { + inflightIndex.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId, entryId + 1); + } + + private boolean untrackInflight(long ledgerId, long entryId) { + LongBitmap bitSet = inflightIndex.get(ledgerId); + if (bitSet == null || !bitSet.contains(entryId, entryId + 1)) { + return false; + } + bitSet.remove(entryId, entryId + 1); + if (bitSet.isEmpty()) { + inflightIndex.remove(ledgerId); } + return true; + } - return Optional.ofNullable(immutableBuckets.get(ledgerId)); + private boolean inflightContains(long ledgerId, long entryId) { + LongBitmap bitSet = inflightIndex.get(ledgerId); + return bitSet != null && bitSet.contains(entryId, entryId + 1); } private void afterCreateImmutableBucket(Pair immutableBucketDelayedIndexPair, long startTime) { if (immutableBucketDelayedIndexPair != null) { ImmutableBucket immutableBucket = immutableBucketDelayedIndexPair.getLeft(); - putBucket(Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId), + putBucket(Range.closed(immutableBucket.getStartLedgerId(), immutableBucket.getEndLedgerId()), immutableBucket); + immutableBucket.getDelayedIndexBitMap().forEach((ledgerId, bitmap) -> + bitmap.forEachLong(entryId -> untrackInflight(ledgerId, entryId))); + DelayedIndex lastDelayedIndex = immutableBucketDelayedIndexPair.getRight(); snapshotSegmentLastIndexMap.put( new SnapshotKey(lastDelayedIndex.getLedgerId(), lastDelayedIndex.getEntryId()), @@ -394,9 +422,9 @@ private void afterCreateImmutableBucket(Pair immu immutableBucket.setSnapshotSegments(null); }); - immutableBucket.setCurrentSegmentEntryId(immutableBucket.lastSegmentEntryId); + immutableBucket.setCurrentSegmentEntryId(immutableBucket.getLastSegmentEntryId()); removeBucket( - Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId)); + Range.closed(immutableBucket.getStartLedgerId(), immutableBucket.getEndLedgerId())); snapshotSegmentLastIndexMap.remove( new SnapshotKey(lastDelayedIndex.getLedgerId(), lastDelayedIndex.getEntryId())); } @@ -417,7 +445,7 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver return true; } - boolean existBucket = findImmutableBucket(ledgerId).isPresent(); + boolean existBucket = findImmutableBucket(ledgerId) != null; // Create bucket snapshot if (!existBucket && ledgerId > lastMutableBucket.endLedgerId @@ -451,8 +479,8 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver // Message index belongs to previous bucket range or the current mutable bucket range, // enter sharedBucketPriorityQueue directly sharedBucketPriorityQueue.add(deliverAt, ledgerId, entryId); - lastMutableBucket.putIndexBit(ledgerId, entryId); } + trackInflight(ledgerId, entryId); numberDelayedMessages.incrementAndGet(); log.debug() @@ -475,15 +503,15 @@ private synchronized List selectMergedBuckets(final List { // We should skip the bucket which last segment already been load to memory, // avoid record replicated index. - return bucket.lastSegmentEntryId > bucket.currentSegmentEntryId && !bucket.merging; + return bucket.getLastSegmentEntryId() > bucket.getCurrentSegmentEntryId() && !bucket.merging; })) { long numberMessages = immutableBuckets.stream() - .mapToLong(bucket -> bucket.numberBucketDelayedMessages) + .mapToLong(bucket -> bucket.getNumberBucketDelayedMessages()) .sum(); if (numberMessages <= minNumberMessages) { minNumberMessages = numberMessages; long scheduleTimestamp = immutableBuckets.stream() - .mapToLong(bucket -> bucket.firstScheduleTimestamps.get(bucket.currentSegmentEntryId + 1)) + .mapToLong(bucket -> bucket.getFirstScheduleTimestamps().get(bucket.getCurrentSegmentEntryId() + 1)) .min().getAsLong(); if (scheduleTimestamp < minScheduleTimestamp) { minScheduleTimestamp = scheduleTimestamp; @@ -514,7 +542,7 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot() { return CompletableFuture.completedFuture(null); } - final String bucketsStr = toBeMergeImmutableBuckets.stream().map(Bucket::bucketKey).collect( + final String bucketsStr = toBeMergeImmutableBuckets.stream().map(ImmutableBucket::bucketKey).collect( Collectors.joining(",")).replaceAll(DELAYED_BUCKET_KEY_PREFIX + "_", ""); log.info() .attr("bucketKeys", bucketsStr) @@ -577,14 +605,14 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List delayedIndexBitMap = - new HashMap<>(buckets.get(0).getDelayedIndexBitMap()); + Long2ObjectMap delayedIndexBitMap = + new Long2ObjectOpenHashMap<>(buckets.get(0).getDelayedIndexBitMap()); for (int i = 1; i < buckets.size(); i++) { - buckets.get(i).delayedIndexBitMap.forEach((ledgerId, bitMapB) -> { + buckets.get(i).getDelayedIndexBitMap().forEach((ledgerId, bitMapB) -> { delayedIndexBitMap.compute(ledgerId, (k, bitMap) -> { if (bitMap == null) { return bitMapB; @@ -609,7 +637,7 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List getScheduledMessages(int maxMessages) break; } - final int preSegmentEntryId = bucket.currentSegmentEntryId; + final int preSegmentEntryId = bucket.getCurrentSegmentEntryId(); log.debug() .attr("bucketKey", bucket.bucketKey()) .attr("nextSegmentEntryId", preSegmentEntryId + 1) @@ -717,7 +745,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages) synchronized (BucketDelayedDeliveryTracker.this) { this.snapshotSegmentLastIndexMap.remove(snapshotKey); if (CollectionUtils.isEmpty(indexList)) { - removeBucket(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); + removeBucket(Range.closed(bucket.getStartLedgerId(), bucket.getEndLedgerId())); bucket.asyncDeleteBucketSnapshot(stats); return; } @@ -747,7 +775,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages) log.info() .attr("bucketKey", bucket.bucketKey()) .attr("segmentEntryId", - (preSegmentEntryId == bucket.lastSegmentEntryId) ? "-1" : preSegmentEntryId + (preSegmentEntryId == bucket.getLastSegmentEntryId()) ? "-1" : preSegmentEntryId + 1) .log("Load next bucket snapshot segment finish"); @@ -804,6 +832,7 @@ public synchronized CompletableFuture clear() { synchronized (BucketDelayedDeliveryTracker.this) { CompletableFuture future = cleanImmutableBuckets(); sharedBucketPriorityQueue.clear(); + inflightIndex.clear(); lastMutableBucket.clear(); snapshotSegmentLastIndexMap.clear(); numberDelayedMessages.set(0); @@ -851,21 +880,21 @@ private CompletableFuture cleanImmutableBuckets() { } private boolean removeIndexBit(long ledgerId, long entryId) { - if (lastMutableBucket.removeIndexBit(ledgerId, entryId)) { + if (untrackInflight(ledgerId, entryId)) { return true; } - return findImmutableBucket(ledgerId).map(bucket -> bucket.removeIndexBit(ledgerId, entryId)) - .orElse(false); + ImmutableBucket bucket = findImmutableBucket(ledgerId); + return bucket != null && bucket.removeIndexBit(ledgerId, entryId); } public synchronized boolean containsMessage(long ledgerId, long entryId) { - if (lastMutableBucket.containsMessage(ledgerId, entryId)) { + if (inflightContains(ledgerId, entryId)) { return true; } - return findImmutableBucket(ledgerId).map(bucket -> bucket.containsMessage(ledgerId, entryId)) - .orElse(false); + ImmutableBucket bucket = findImmutableBucket(ledgerId); + return bucket != null && bucket.containsMessage(ledgerId, entryId); } public Map genTopicMetricMap() { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java index 544cedce497ae..9634d2a8495fb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java @@ -19,9 +19,12 @@ package org.apache.pulsar.broker.delayed.bucket; import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry; +import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.DELAYED_BUCKET_KEY_PREFIX; import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.NULL_LONG_PROMISE; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -29,19 +32,36 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Supplier; import lombok.CustomLog; +import lombok.Getter; import lombok.Setter; -import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.mutable.MutableLong; import org.apache.pulsar.broker.delayed.proto.DelayedIndex; import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata; import org.apache.pulsar.broker.delayed.proto.SnapshotSegment; +import org.apache.pulsar.common.util.Codec; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.LongBitmap; import org.apache.pulsar.common.util.collections.LongBitmaps; @CustomLog -class ImmutableBucket extends Bucket { +class ImmutableBucket { + + static final String DELIMITER = "_"; + static final int MaxRetryTimes = 3; + + private final BucketContext ctx; + + @Getter + private final long startLedgerId; + + @Getter + private final long endLedgerId; + + @Getter + @Setter + private Long2ObjectMap delayedIndexBitMap = new Long2ObjectOpenHashMap<>(); @Setter private List snapshotSegments; @@ -49,11 +69,135 @@ class ImmutableBucket extends Bucket { boolean merging = false; @Setter + @Getter List firstScheduleTimestamps = new ArrayList<>(); - ImmutableBucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, - BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { - super(dispatcherName, cursor, sequencer, storage, startLedgerId, endLedgerId); + @Getter + @Setter + private long numberBucketDelayedMessages; + + @Getter + @Setter + private int lastSegmentEntryId; + + @Getter + @Setter + private volatile int currentSegmentEntryId; + + @Getter + @Setter + private volatile long snapshotLength; + + @Getter + @Setter + private volatile Long bucketId; + + @Getter + @Setter + private volatile CompletableFuture snapshotCreateFuture; + + boolean containsMessage(long ledgerId, long entryId) { + LongBitmap bitSet = delayedIndexBitMap.get(ledgerId); + if (bitSet == null) { + return false; + } + return bitSet.contains(entryId); + } + + boolean removeIndexBit(long ledgerId, long entryId) { + LongBitmap bitSet = delayedIndexBitMap.get(ledgerId); + if (bitSet == null) { + return false; + } + boolean contained = bitSet.remove(entryId); + if (contained) { + if (bitSet.isEmpty()) { + delayedIndexBitMap.remove(ledgerId); + } + + if (numberBucketDelayedMessages > 0) { + numberBucketDelayedMessages--; + } + } + return contained; + } + + ImmutableBucket(BucketContext ctx, long startLedgerId, long endLedgerId) { + this.ctx = ctx; + this.startLedgerId = startLedgerId; + this.endLedgerId = endLedgerId; + } + + String bucketKey() { + return String.join(DELIMITER, DELAYED_BUCKET_KEY_PREFIX, String.valueOf(startLedgerId), + String.valueOf(endLedgerId)); + } + + Optional> getSnapshotCreateFuture() { + return Optional.ofNullable(snapshotCreateFuture); + } + + Optional getBucketId() { + return Optional.ofNullable(bucketId); + } + + long getAndUpdateBucketId() { + Optional bucketIdOptional = getBucketId(); + if (bucketIdOptional.isPresent()) { + return bucketIdOptional.get(); + } + + String bucketIdStr = ctx.cursor().getCursorProperties().get(bucketKey()); + long bucketId = Long.parseLong(bucketIdStr); + setBucketId(bucketId); + return bucketId; + } + + CompletableFuture asyncSaveBucketSnapshot( + SnapshotMetadata snapshotMetadata, List bucketSnapshotSegments) { + final String bucketKey = bucketKey(); + final String cursorName = Codec.decode(ctx.cursor().getName()); + final String dispatcher = ctx.dispatcherName(); + final String topicName = dispatcher.substring(0, dispatcher.lastIndexOf(" / " + cursorName)); + return executeWithRetry( + () -> ctx.bucketSnapshotStorage().createBucketSnapshot(snapshotMetadata, bucketSnapshotSegments, + bucketKey, topicName, cursorName) + .whenComplete((__, ex) -> { + if (ex != null) { + log.warn() + .attr("dispatcher", dispatcher) + .attr("bucketKey", bucketKey) + .exception(ex) + .log("Failed to create bucket snapshot"); + } + }), BucketSnapshotPersistenceException.class, MaxRetryTimes).thenCompose(newBucketId -> { + setBucketId(newBucketId); + + return putBucketKeyId(bucketKey, newBucketId).exceptionally(ex -> { + log.warn() + .attr("dispatcher", dispatcher) + .attr("bucketKey", bucketKey) + .attr("bucketId", newBucketId) + .exception(ex) + .log("Failed to record bucketId to cursor property"); + return null; + }).thenApply(__ -> newBucketId); + }); + } + + private CompletableFuture putBucketKeyId(String bucketKey, Long bucketId) { + if (bucketId == null) { + return FutureUtil.failedFuture(new NullPointerException("Expected bucketId should not be null")); + } + return ctx.sequencer().sequential(() -> + executeWithRetry(() -> ctx.cursor().putCursorProperty(bucketKey, String.valueOf(bucketId)), + ManagedLedgerException.BadVersionException.class, MaxRetryTimes)); + } + + CompletableFuture removeBucketCursorProperty(String bucketKey) { + return ctx.sequencer().sequential(() -> + executeWithRetry(() -> ctx.cursor().removeCursorProperty(bucketKey), + ManagedLedgerException.BadVersionException.class, MaxRetryTimes)); } public Optional> getSnapshotSegments() { @@ -76,11 +220,11 @@ private CompletableFuture> asyncLoadNextBucketSnapshotEntry(b final long cutoffTime = cutoffTimeSupplier.get(); // Load Metadata of bucket snapshot final String bucketKey = bucketKey(); - loadMetaDataFuture = executeWithRetry(() -> bucketSnapshotStorage.getBucketSnapshotMetadata(bucketId) + loadMetaDataFuture = executeWithRetry(() -> ctx.bucketSnapshotStorage().getBucketSnapshotMetadata(bucketId) .whenComplete((___, ex) -> { if (ex != null) { log.warn() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketKey", bucketKey) .attr("bucketId", bucketId) .exception(ex) @@ -119,11 +263,11 @@ private CompletableFuture> asyncLoadNextBucketSnapshotEntry(b } return executeWithRetry( - () -> bucketSnapshotStorage.getBucketSnapshotSegment(bucketId, nextSegmentEntryId, + () -> ctx.bucketSnapshotStorage().getBucketSnapshotSegment(bucketId, nextSegmentEntryId, nextSegmentEntryId).whenComplete((___, ex) -> { if (ex != null) { log.warn() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketKey", bucketKey()) .attr("bucketId", bucketId) .attr("segmentEntryId", nextSegmentEntryId) @@ -184,11 +328,11 @@ CompletableFuture> getRemainSnapshotSegment() { return CompletableFuture.completedFuture(Collections.emptyList()); } return executeWithRetry(() -> { - return bucketSnapshotStorage.getBucketSnapshotSegment(getAndUpdateBucketId(), nextSegmentEntryId, + return ctx.bucketSnapshotStorage().getBucketSnapshotSegment(getAndUpdateBucketId(), nextSegmentEntryId, lastSegmentEntryId).whenComplete((__, ex) -> { if (ex != null) { log.warn() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketKey", bucketKey()) .attr("nextSegmentEntryId", nextSegmentEntryId) .attr("lastSegmentEntryId", lastSegmentEntryId) @@ -205,12 +349,12 @@ CompletableFuture asyncDeleteBucketSnapshot(BucketDelayedMessageIndexStats String bucketKey = bucketKey(); long bucketId = getAndUpdateBucketId(); - return executeWithRetry(() -> bucketSnapshotStorage.deleteBucketSnapshot(bucketId), + return executeWithRetry(() -> ctx.bucketSnapshotStorage().deleteBucketSnapshot(bucketId), BucketSnapshotPersistenceException.class, MaxRetryTimes) .whenComplete((__, ex) -> { if (ex != null) { log.error() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketId", bucketId) .attr("bucketKey", bucketKey) .exception(ex) @@ -219,7 +363,7 @@ CompletableFuture asyncDeleteBucketSnapshot(BucketDelayedMessageIndexStats stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.delete); } else { log.info() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketId", bucketId) .attr("bucketKey", bucketKey) .log("Delete bucket snapshot finish"); @@ -239,10 +383,10 @@ CompletableFuture clear(BucketDelayedMessageIndexStats stats) { protected CompletableFuture asyncUpdateSnapshotLength() { long bucketId = getAndUpdateBucketId(); - return bucketSnapshotStorage.getBucketSnapshotLength(bucketId).whenComplete((length, ex) -> { + return ctx.bucketSnapshotStorage().getBucketSnapshotLength(bucketId).whenComplete((length, ex) -> { if (ex != null) { log.error() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketId", bucketId) .attr("bucketKey", bucketKey()) .exception(ex) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java index ff0d4f5f19859..046653c0dfa88 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java @@ -19,32 +19,35 @@ package org.apache.pulsar.broker.delayed.bucket; import static com.google.common.base.Preconditions.checkArgument; +import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; import java.util.ArrayList; -import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import lombok.CustomLog; -import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.delayed.proto.DelayedIndex; import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata; import org.apache.pulsar.broker.delayed.proto.SnapshotSegment; import org.apache.pulsar.broker.delayed.proto.SnapshotSegmentMetadata; -import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.LongBitmap; import org.apache.pulsar.common.util.collections.LongBitmaps; import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; @CustomLog -class MutableBucket extends Bucket implements AutoCloseable { +class MutableBucket implements AutoCloseable { + + private final BucketContext ctx; private final TripleLongPriorityQueue priorityQueue; - MutableBucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, - BucketSnapshotStorage bucketSnapshotStorage) { - super(dispatcherName, cursor, sequencer, bucketSnapshotStorage, -1L, -1L); + long startLedgerId = -1L; + long endLedgerId = -1L; + + MutableBucket(BucketContext ctx) { + this.ctx = ctx; this.priorityQueue = new TripleLongPriorityQueue(); } @@ -62,7 +65,7 @@ Pair createImmutableBucketAndAsyncPersistent( TripleLongPriorityQueue sharedQueue, DelayedIndexQueue delayedIndexQueue, final long startLedgerId, final long endLedgerId) { log.debug() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("startLedgerId", startLedgerId) .attr("endLedgerId", endLedgerId) .log("Creating bucket snapshot"); @@ -73,9 +76,9 @@ Pair createImmutableBucketAndAsyncPersistent( List bucketSnapshotSegments = new ArrayList<>(); List segmentMetadataList = new ArrayList<>(); - Map immutableBucketBitMap = new HashMap<>(); + Long2ObjectMap immutableBucketBitMap = new Long2ObjectOpenHashMap<>(); - Map bitMap = new HashMap<>(); + Long2ObjectMap bitMap = new Long2ObjectOpenHashMap<>(); SnapshotSegment snapshotSegment = new SnapshotSegment(); SnapshotSegmentMetadata segmentMetadata = new SnapshotSegmentMetadata(); @@ -96,8 +99,6 @@ Pair createImmutableBucketAndAsyncPersistent( final long ledgerId = delayedIndex.getLedgerId(); final long entryId = delayedIndex.getEntryId(); - removeIndexBit(ledgerId, entryId); - checkArgument(ledgerId >= startLedgerId && ledgerId <= endLedgerId); // Move first segment of bucket snapshot to sharedBucketPriorityQueue @@ -147,8 +148,7 @@ Pair createImmutableBucketAndAsyncPersistent( final int lastSegmentEntryId = segmentMetadataList.size(); - ImmutableBucket bucket = new ImmutableBucket(dispatcherName, cursor, sequencer, bucketSnapshotStorage, - startLedgerId, endLedgerId); + ImmutableBucket bucket = new ImmutableBucket(ctx, startLedgerId, endLedgerId); bucket.setCurrentSegmentEntryId(1); bucket.setNumberBucketDelayedMessages(numMessages); bucket.setLastSegmentEntryId(lastSegmentEntryId); @@ -165,7 +165,7 @@ Pair createImmutableBucketAndAsyncPersistent( DelayedIndex lastDelayedIndex = firstSnapshotSegment.getIndexeAt(firstSnapshotSegment.getIndexesCount() - 1); Pair result = Pair.of(bucket, lastDelayedIndex); - CompletableFuture future = asyncSaveBucketSnapshot(bucket, + CompletableFuture future = bucket.asyncSaveBucketSnapshot( bucketSnapshotMetadata, bucketSnapshotSegments); bucket.setSnapshotCreateFuture(future); @@ -194,7 +194,6 @@ void resetLastMutableBucketRange() { void clear() { this.resetLastMutableBucketRange(); - this.delayedIndexBitMap.clear(); this.priorityQueue.clear(); } @@ -224,6 +223,5 @@ void addMessage(long ledgerId, long entryId, long deliverAt) { this.startLedgerId = ledgerId; } this.endLedgerId = ledgerId; - putIndexBit(ledgerId, entryId); } } From 005ac963e7937421b0b5f8cf31113a12caa94c1c Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Mon, 20 Jul 2026 18:30:07 +0800 Subject: [PATCH 2/3] Fix inflightIndex check --- .../delayed/bucket/BucketDelayedDeliveryTracker.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java index a8b1fd3dc9b4a..896859d50f177 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java @@ -349,15 +349,15 @@ private ImmutableBucket findImmutableBucket(long ledgerId) { } private void trackInflight(long ledgerId, long entryId) { - inflightIndex.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId, entryId + 1); + inflightIndex.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId); } private boolean untrackInflight(long ledgerId, long entryId) { LongBitmap bitSet = inflightIndex.get(ledgerId); - if (bitSet == null || !bitSet.contains(entryId, entryId + 1)) { + if (bitSet == null || !bitSet.contains(entryId)) { return false; } - bitSet.remove(entryId, entryId + 1); + bitSet.remove(entryId); if (bitSet.isEmpty()) { inflightIndex.remove(ledgerId); } @@ -366,7 +366,7 @@ private boolean untrackInflight(long ledgerId, long entryId) { private boolean inflightContains(long ledgerId, long entryId) { LongBitmap bitSet = inflightIndex.get(ledgerId); - return bitSet != null && bitSet.contains(entryId, entryId + 1); + return bitSet != null && bitSet.contains(entryId); } private void afterCreateImmutableBucket(Pair immutableBucketDelayedIndexPair, From 7303c6f59a83c21ad20a4248a7a72f414060cd06 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Mon, 20 Jul 2026 19:55:37 +0800 Subject: [PATCH 3/3] Update BucketDelayedDeliveryTracker.java --- .../broker/delayed/bucket/BucketDelayedDeliveryTracker.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java index 896859d50f177..f8fdca01a8a1e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java @@ -511,7 +511,8 @@ private synchronized List selectMergedBuckets(final List bucket.getFirstScheduleTimestamps().get(bucket.getCurrentSegmentEntryId() + 1)) + .mapToLong(bucket -> bucket.getFirstScheduleTimestamps() + .get(bucket.getCurrentSegmentEntryId() + 1)) .min().getAsLong(); if (scheduleTimestamp < minScheduleTimestamp) { minScheduleTimestamp = scheduleTimestamp;