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 1bcfcd5eb986e..e5ddd0a6d4c1a 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; @@ -53,7 +54,6 @@ import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.collections4.MapUtils; -import org.apache.commons.lang3.mutable.MutableLong; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.DelayedDeliveryContext; @@ -96,7 +96,9 @@ public static record SnapshotKey(long ledgerId, long entryId) {} private final int maxNumBuckets; - private final AtomicLong numberDelayedMessages = new AtomicLong(0); + @Getter + @VisibleForTesting + private final BucketContext ctx; @Getter @VisibleForTesting @@ -110,6 +112,10 @@ public static record SnapshotKey(long ledgerId, long entryId) {} @VisibleForTesting private final RangeMap immutableBuckets; + @Getter + @VisibleForTesting + private final BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + @Getter @VisibleForTesting private final AtomicLong bucketsCount = new AtomicLong(0); @@ -167,15 +173,14 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, this.sharedBucketPriorityQueue = new TripleLongPriorityQueue(); this.immutableBuckets = TreeRangeMap.create(); 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. try { - long recoveredMessages = recoverBucketSnapshot(); - this.numberDelayedMessages.set(recoveredMessages); + recoverBucketSnapshot(); } catch (RecoverDelayedDeliveryTrackerException e) { close(); throw e; @@ -183,24 +188,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); } }); @@ -257,10 +259,9 @@ private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryT immutableBucket.asyncDeleteBucketSnapshot(stats); } - MutableLong numberDelayedMessages = new MutableLong(0); long totalLength = 0; for (ImmutableBucket bucket : immutableBucketMap.values()) { - numberDelayedMessages.add(bucket.numberBucketDelayedMessages); + index.restore(bucket.getDelayedIndexBitMap()); totalLength += bucket.getSnapshotLength(); } totalSnapshotLengthBytes.set(totalLength); @@ -268,10 +269,10 @@ private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryT log.info() .attr("buckets", immutableBucketMap.size()) - .attr("numberDelayedMessages", numberDelayedMessages.longValue()) + .attr("numberDelayedMessages", index.size()) .log("Recover delayed message index bucket snapshot finish"); - return numberDelayedMessages.longValue(); + return index.size(); } /** @@ -309,7 +310,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,19 +337,15 @@ public void run(Timeout timeout) throws Exception { super.run(timeout); } - private Optional findImmutableBucket(long ledgerId) { - if (immutableBuckets.asMapOfRanges().isEmpty()) { - return Optional.empty(); - } - - return Optional.ofNullable(immutableBuckets.get(ledgerId)); + private ImmutableBucket findImmutableBucket(long ledgerId) { + return immutableBuckets.get(ledgerId); } 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); DelayedIndex lastDelayedIndex = immutableBucketDelayedIndexPair.getRight(); @@ -394,9 +391,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())); } @@ -410,6 +407,7 @@ private void afterCreateImmutableBucket(Pair immu @Override public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) { if (deliverAt < 0 || deliverAt <= getCutoffTime()) { + removeIndexBit(ledgerId, entryId); return false; } @@ -417,7 +415,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,10 +449,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); } - - numberDelayedMessages.incrementAndGet(); + index.track(ledgerId, entryId); log.debug() .attr("ledgerId", ledgerId) .attr("entryId", entryId) @@ -479,10 +475,10 @@ 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; @@ -490,7 +486,8 @@ synchronized List selectMergedBuckets(final List bucket.firstScheduleTimestamps.get(bucket.currentSegmentEntryId)) + .mapToLong(bucket -> bucket.getFirstScheduleTimestamps() + .get(bucket.getCurrentSegmentEntryId())) .min().getAsLong(); if (scheduleTimestamp < minScheduleTimestamp) { minScheduleTimestamp = scheduleTimestamp; @@ -521,7 +518,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) @@ -567,13 +564,13 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List>> getRemainFutures = - buckets.stream().map(ImmutableBucket::getRemainSnapshotSegment).toList(); + List>> getAllSnapshotFutures = + buckets.stream().map(ImmutableBucket::getAllSnapshotSegments).toList(); - return FutureUtil.waitForAll(getRemainFutures) + return FutureUtil.waitForAll(getAllSnapshotFutures) .thenApply(__ -> { return CombinedSegmentDelayedIndexQueue.wrap( - getRemainFutures.stream().map(CompletableFuture::join).toList()); + getAllSnapshotFutures.stream().map(CompletableFuture::join).toList()); }) .thenAccept(combinedDelayedIndexQueue -> { synchronized (BucketDelayedDeliveryTracker.this) { @@ -584,14 +581,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; @@ -616,7 +613,7 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List getScheduledMessages(int maxMessages) long entryId = sharedBucketPriorityQueue.peekN3(); if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) { sharedBucketPriorityQueue.pop(); - if (removeIndexBit(ledgerId, entryId)) { - numberDelayedMessages.decrementAndGet(); - } + removeIndexBit(ledgerId, entryId); continue; } if (timestamp > cutoffTime) { @@ -703,7 +698,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages) break; } - final int preSegmentEntryId = bucket.currentSegmentEntryId; + final int preSegmentEntryId = bucket.getCurrentSegmentEntryId(); log.debug() .attr("bucketKey", bucket.bucketKey()) .attr("nextSegmentEntryId", preSegmentEntryId + 1) @@ -724,7 +719,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; } @@ -754,7 +749,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"); @@ -769,13 +764,13 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages) } } - positions.add(PositionFactory.create(ledgerId, entryId)); - sharedBucketPriorityQueue.pop(); - removeIndexBit(ledgerId, entryId); - - --n; - numberDelayedMessages.decrementAndGet(); + // Dedup: queue may carry the same position twice (initial seal + merge); only the + // first delivery of each position decrements the counter via removeIndexBit. + if (removeIndexBit(ledgerId, entryId)) { + positions.add(PositionFactory.create(ledgerId, entryId)); + --n; + } } updateTimer(); @@ -811,9 +806,9 @@ public synchronized CompletableFuture clear() { synchronized (BucketDelayedDeliveryTracker.this) { CompletableFuture future = cleanImmutableBuckets(); sharedBucketPriorityQueue.clear(); + index.clear(); lastMutableBucket.clear(); snapshotSegmentLastIndexMap.clear(); - numberDelayedMessages.set(0); return future; } }); @@ -850,7 +845,6 @@ private CompletableFuture cleanImmutableBuckets() { List> futures = new ArrayList<>(); bucketsToDelete.forEach((range, bucket) -> { removeBucket(range); - numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages()); futures.add(bucket.clear(stats)); }); @@ -858,21 +852,11 @@ private CompletableFuture cleanImmutableBuckets() { } private boolean removeIndexBit(long ledgerId, long entryId) { - if (lastMutableBucket.removeIndexBit(ledgerId, entryId)) { - return true; - } - - return findImmutableBucket(ledgerId).map(bucket -> bucket.removeIndexBit(ledgerId, entryId)) - .orElse(false); + return index.untrack(ledgerId, entryId); } public synchronized boolean containsMessage(long ledgerId, long entryId) { - if (lastMutableBucket.containsMessage(ledgerId, entryId)) { - return true; - } - - return findImmutableBucket(ledgerId).map(bucket -> bucket.containsMessage(ledgerId, entryId)) - .orElse(false); + return index.contains(ledgerId, entryId); } public Map genTopicMetricMap() { @@ -929,7 +913,8 @@ private CompletableFuture deleteBucketSnapshot(String ledgerName, synchronized (this) { snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket); removeBucket(range); - numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages()); + bucket.getDelayedIndexBitMap().forEach((ledgerId, bitmap) -> + bitmap.forEachLong(entryId -> index.untrack(ledgerId, entryId))); } return null; }); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndex.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndex.java new file mode 100644 index 0000000000000..86bb88e4b6ef8 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndex.java @@ -0,0 +1,83 @@ +/* + * 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 it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; +import javax.annotation.concurrent.ThreadSafe; +import org.apache.pulsar.common.util.collections.LongBitmap; +import org.apache.pulsar.common.util.collections.LongBitmaps; + +/** + * Runtime truth for delayed messages that have been accepted but not yet delivered. + * Co-locates the bitmap and its cardinality so the counter is an invariant of the bitmap + * rather than a discipline callers must maintain. {@link ImmutableBucket#delayedIndexBitMap} + * is a frozen snapshot for BookKeeper writes/merge and is intentionally not consulted here. + */ +@ThreadSafe +final class BucketDelayedMessageIndex { + + private final Long2ObjectMap inflightIndex = new Long2ObjectOpenHashMap<>(); + private final AtomicLong size = new AtomicLong(0); + + /** Idempotent: re-tracking a position already in the index is a no-op. */ + void track(long ledgerId, long entryId) { + if (inflightIndex.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).checkedAdd(entryId)) { + size.incrementAndGet(); + } + } + + /** @return true if the bit was present and removed; false if it was already absent. */ + boolean untrack(long ledgerId, long entryId) { + LongBitmap bitSet = inflightIndex.get(ledgerId); + if (bitSet == null || !bitSet.contains(entryId)) { + return false; + } + bitSet.remove(entryId); + if (bitSet.isEmpty()) { + inflightIndex.remove(ledgerId); + } + size.decrementAndGet(); + return true; + } + + boolean contains(long ledgerId, long entryId) { + LongBitmap bitSet = inflightIndex.get(ledgerId); + return bitSet != null && bitSet.contains(entryId); + } + + long size() { + return size.get(); + } + + void clear() { + inflightIndex.clear(); + size.set(0); + } + + /** + * Bulk-load after recovery. Built on {@link #track} so overlapping bits merge, not double-count. + */ + void restore(Map snapshot) { + snapshot.forEach((ledgerId, bitmap) -> + bitmap.forEachLong(entryId -> track(ledgerId, entryId))); + } +} 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..925c4756019c7 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,29 +19,49 @@ 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.Long2ObjectOpenHashMap; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Optional; 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 Map delayedIndexBitMap = new Long2ObjectOpenHashMap<>(); @Setter private List snapshotSegments; @@ -49,11 +69,109 @@ 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; + + 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 +194,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 +237,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) @@ -178,22 +296,20 @@ private void recoverDelayedIndexBitMapAndNumber(int startSnapshotIndex, setNumberBucketDelayedMessages(numberMessages.longValue()); } - CompletableFuture> getRemainSnapshotSegment() { - int nextSegmentEntryId = currentSegmentEntryId + 1; - if (nextSegmentEntryId > lastSegmentEntryId) { + CompletableFuture> getAllSnapshotSegments() { + if (lastSegmentEntryId < 1) { return CompletableFuture.completedFuture(Collections.emptyList()); } return executeWithRetry(() -> { - return bucketSnapshotStorage.getBucketSnapshotSegment(getAndUpdateBucketId(), nextSegmentEntryId, + return ctx.bucketSnapshotStorage().getBucketSnapshotSegment(getAndUpdateBucketId(), 1, lastSegmentEntryId).whenComplete((__, ex) -> { if (ex != null) { log.warn() - .attr("dispatcher", dispatcherName) + .attr("dispatcher", ctx.dispatcherName()) .attr("bucketKey", bucketKey()) - .attr("nextSegmentEntryId", nextSegmentEntryId) .attr("lastSegmentEntryId", lastSegmentEntryId) .exception(ex) - .log("Failed to get remain bucket snapshot segment"); + .log("Failed to get all bucket snapshot segments for merge"); } }); }, BucketSnapshotPersistenceException.class, MaxRetryTimes); @@ -205,12 +321,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 +335,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 +355,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..0d3630987af8c 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 @@ -26,25 +26,27 @@ 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 +64,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"); @@ -96,8 +98,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 +147,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 +164,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 +193,6 @@ void resetLastMutableBucketRange() { void clear() { this.resetLastMutableBucketRange(); - this.delayedIndexBitMap.clear(); this.priorityQueue.clear(); } @@ -224,6 +222,5 @@ void addMessage(long ledgerId, long entryId, long deliverAt) { this.startLedgerId = ledgerId; } this.endLedgerId = ledgerId; - putIndexBit(ledgerId, entryId); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java index 28626496e49ae..2b5cd712a429d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java @@ -149,7 +149,8 @@ public Object[][] provider(Method method) throws Exception { new BucketDelayedDeliveryTracker(dispatcher, timer, 500, clock, true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50) }}; - case "testMergeSnapshot", "testWithBkException", "testWithCreateFailDowngrade" -> new Object[][]{{ + case "testMergeSnapshot", "testWithBkException", "testWithCreateFailDowngrade", + "testMergePreservesAllSnapshotSegments" -> new Object[][]{{ new BucketDelayedDeliveryTracker(dispatcher, timer, 100000, clock, true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 10) }}; @@ -161,7 +162,8 @@ public Object[][] provider(Method method) throws Exception { new BucketDelayedDeliveryTracker(dispatcher, timer, 100000, clock, true, bucketSnapshotStorage, 1000, TimeUnit.MILLISECONDS.toMillis(100), -1, 50) }}; - case "testExpiredTrackedMessageReturnsFalse", "testRecoverThenExpireAddMessage" -> new Object[][]{{ + case "testExpiredTrackedMessageReturnsFalse", "testRecoverThenExpireAddMessage", + "testExpiredTrackedMessageDecrementsCount" -> new Object[][]{{ new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock, true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50) }}; @@ -241,6 +243,40 @@ public void testRecoverThenExpireAddMessage(BucketDelayedDeliveryTracker tracker tracker2.close(); } + @Test(dataProvider = "delayedTracker") + public void testExpiredTrackedMessageDecrementsCount(BucketDelayedDeliveryTracker tracker) { + clockTime.set(1000); + tracker.addMessage(1, 1, 2000); + assertEquals(tracker.getNumberOfDelayedMessages(), 1); + + clockTime.set(2500); + assertFalse(tracker.addMessage(1, 1, 2000)); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + assertFalse(tracker.containsMessage(1, 1)); + tracker.close(); + } + + @Test(dataProvider = "delayedTracker") + public void testMergePreservesAllSnapshotSegments(BucketDelayedDeliveryTracker tracker) throws Exception { + clockTime.set(0); + for (int i = 1; i <= 56; i++) { + tracker.addMessage(i, i, i * 10); + } + Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().values().stream() + .noneMatch(x -> x.merging))); + assertEquals(tracker.getNumberOfDelayedMessages(), 56); + + tracker.close(); + clockTime.set(0); + BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker( + dispatcher, timer, 100000, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 10); + + assertEquals(tracker2.getNumberOfDelayedMessages(), 55); + tracker2.close(); + } + @Test(dataProvider = "delayedTracker", invocationCount = 10) public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) throws Exception { for (int i = 1; i <= 100; i++) { @@ -366,7 +402,7 @@ public void testMergeSnapshot(final BucketDelayedDeliveryTracker tracker) throws clockTime.set(110 * 10); NavigableSet scheduledMessages = new TreeSet<>(); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { scheduledMessages.addAll(tracker2.getScheduledMessages(110)); assertEquals(scheduledMessages.size(), 110); }); @@ -443,7 +479,7 @@ public void testWithBkException(final BucketDelayedDeliveryTracker tracker) thro assertEquals(tracker2.getScheduledMessages(100).size(), 0); Set scheduledMessages = new TreeSet<>(); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { scheduledMessages.addAll(tracker2.getScheduledMessages(100)); assertEquals(scheduledMessages.size(), delayedMessagesInSnapshotValue); }); @@ -568,9 +604,7 @@ public CompletableFuture deleteBucketSnapshot(long bucketId) { private ImmutableBucket createMergeableBucket(TrackerWithStorage trackerWithStorage, long startLedgerId, long endLedgerId, List firstScheduleTimestamps) { - MutableBucket mutableBucket = trackerWithStorage.tracker.getLastMutableBucket(); - ImmutableBucket bucket = new ImmutableBucket(mutableBucket.dispatcherName, mutableBucket.cursor, - mutableBucket.sequencer, mutableBucket.bucketSnapshotStorage, startLedgerId, endLedgerId); + ImmutableBucket bucket = new ImmutableBucket(trackerWithStorage.tracker.getCtx(), startLedgerId, endLedgerId); bucket.setCurrentSegmentEntryId(1); bucket.setLastSegmentEntryId(firstScheduleTimestamps.size()); bucket.setFirstScheduleTimestamps(firstScheduleTimestamps); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexTest.java new file mode 100644 index 0000000000000..57774605580f1 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexTest.java @@ -0,0 +1,179 @@ +/* + * 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.assertj.core.api.Assertions.assertThat; +import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; +import org.apache.pulsar.common.util.collections.LongBitmap; +import org.apache.pulsar.common.util.collections.LongBitmaps; +import org.testng.annotations.Test; + +public class BucketDelayedMessageIndexTest { + + @Test + public void trackThenContains() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + index.track(7L, 100L); + + assertThat(index.contains(7L, 100L)).isTrue(); + assertThat(index.contains(7L, 101L)).isFalse(); + assertThat(index.contains(8L, 100L)).isFalse(); + assertThat(index.size()).isEqualTo(1L); + } + + @Test + public void untrackReturnsTrueFirstTimeAndFalseAfter() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + index.track(1L, 1L); + + assertThat(index.untrack(1L, 1L)).isTrue(); + assertThat(index.size()).isZero(); + + assertThat(index.untrack(1L, 1L)).isFalse(); + assertThat(index.size()).isZero(); + } + + @Test + public void untrackOnAbsentBitIsSafe() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + assertThat(index.untrack(99L, 99L)).isFalse(); + assertThat(index.size()).isZero(); + assertThat(index.contains(99L, 99L)).isFalse(); + } + + @Test + public void trackIsIdempotent() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + index.track(3L, 5L); + index.track(3L, 5L); + index.track(3L, 5L); + + assertThat(index.size()).isEqualTo(1L); + assertThat(index.contains(3L, 5L)).isTrue(); + } + + @Test + public void trackAcrossManyLedgersKeepsCounterCorrect() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + for (long ledger = 1; ledger <= 5; ledger++) { + for (long entry = 1; entry <= 10; entry++) { + index.track(ledger, entry); + } + } + + assertThat(index.size()).isEqualTo(50L); + + // Drain half. + for (long ledger = 1; ledger <= 5; ledger++) { + for (long entry = 1; entry <= 5; entry++) { + assertThat(index.untrack(ledger, entry)).isTrue(); + } + } + assertThat(index.size()).isEqualTo(25L); + } + + @Test + public void clearResetsBitmapAndCounter() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + index.track(1L, 1L); + index.track(2L, 2L); + assertThat(index.size()).isEqualTo(2L); + + index.clear(); + + assertThat(index.size()).isZero(); + assertThat(index.contains(1L, 1L)).isFalse(); + assertThat(index.contains(2L, 2L)).isFalse(); + + // Index remains usable after clear. + index.track(3L, 3L); + assertThat(index.size()).isEqualTo(1L); + assertThat(index.contains(3L, 3L)).isTrue(); + } + + @Test + public void restoreLoadsBitsFromSnapshot() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + Long2ObjectMap snapshot = new Long2ObjectOpenHashMap<>(); + LongBitmap ledger1 = LongBitmaps.create(); + ledger1.add(10L); + ledger1.add(11L); + snapshot.put(1L, ledger1); + LongBitmap ledger2 = LongBitmaps.create(); + ledger2.add(20L); + snapshot.put(2L, ledger2); + + index.restore(snapshot); + + assertThat(index.size()).isEqualTo(3L); + assertThat(index.contains(1L, 10L)).isTrue(); + assertThat(index.contains(1L, 11L)).isTrue(); + assertThat(index.contains(2L, 20L)).isTrue(); + } + + @Test + public void restoreIsIdempotentOnOverlappingSnapshots() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + + Long2ObjectMap firstBucket = new Long2ObjectOpenHashMap<>(); + LongBitmap bits = LongBitmaps.create(); + bits.add(5L); + bits.add(6L); + firstBucket.put(7L, bits); + + Long2ObjectMap secondBucket = new Long2ObjectOpenHashMap<>(); + LongBitmap overlapping = LongBitmaps.create(); + overlapping.add(5L); // overlap with firstBucket + overlapping.add(8L); + secondBucket.put(7L, overlapping); + + index.restore(firstBucket); + index.restore(secondBucket); + + assertThat(index.size()).isEqualTo(3L); // 5, 6, 8 — not 4 + assertThat(index.contains(7L, 5L)).isTrue(); + assertThat(index.contains(7L, 6L)).isTrue(); + assertThat(index.contains(7L, 8L)).isTrue(); + } + + @Test + public void restoreAfterTrackMergesCorrectly() { + BucketDelayedMessageIndex index = new BucketDelayedMessageIndex(); + index.track(1L, 1L); + assertThat(index.size()).isEqualTo(1L); + + Long2ObjectMap snapshot = new Long2ObjectOpenHashMap<>(); + LongBitmap bits = LongBitmaps.create(); + bits.add(1L); // overlap with the existing tracked bit + bits.add(2L); + snapshot.put(1L, bits); + + index.restore(snapshot); + + assertThat(index.size()).isEqualTo(2L); + assertThat(index.contains(1L, 1L)).isTrue(); + assertThat(index.contains(1L, 2L)).isTrue(); + } +}