This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b61f3d5ca28 [improve][broker] Add LongBitmap abstraction and migrate
RoaringBitmap usage to LongBitmap (#26117)
b61f3d5ca28 is described below
commit b61f3d5ca28df13855eaaa40c1e81e9c834eb814
Author: Zixuan Liu <[email protected]>
AuthorDate: Wed Jul 1 09:43:59 2026 +0800
[improve][broker] Add LongBitmap abstraction and migrate RoaringBitmap
usage to LongBitmap (#26117)
---
.../util/collections/LongBitmapBenchmark.java | 141 ++++
.../delayed/InMemoryDelayedDeliveryTracker.java | 59 +-
.../pulsar/broker/delayed/bucket/Bucket.java | 11 +-
.../bucket/BucketDelayedDeliveryTracker.java | 6 +-
.../broker/delayed/bucket/ImmutableBucket.java | 34 +-
.../broker/delayed/bucket/MutableBucket.java | 22 +-
.../broker/service/ConsumerNameIndexTracker.java | 9 +-
.../broker/service/DrainingHashesTracker.java | 27 +-
pulsar-common/build.gradle.kts | 1 +
.../util/collections/ConcurrentRoaringBitmap.java | 441 +++++++++++
.../pulsar/common/util/collections/LongBitmap.java | 170 +++++
.../common/util/collections/LongBitmaps.java | 61 ++
.../collections/LongBitmapCompatibilityTest.java | 235 ++++++
.../common/util/collections/LongBitmapTest.java | 824 +++++++++++++++++++++
14 files changed, 1942 insertions(+), 99 deletions(-)
diff --git
a/microbench/src/main/java/org/apache/pulsar/common/util/collections/LongBitmapBenchmark.java
b/microbench/src/main/java/org/apache/pulsar/common/util/collections/LongBitmapBenchmark.java
new file mode 100644
index 00000000000..9fa3c464d73
--- /dev/null
+++
b/microbench/src/main/java/org/apache/pulsar/common/util/collections/LongBitmapBenchmark.java
@@ -0,0 +1,141 @@
+/*
+ * 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.common.util.collections;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.roaringbitmap.RoaringBitmap;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+
+/**
+ * JMH benchmark for {@link LongBitmap} ({@link ConcurrentRoaringBitmap}),
compared against
+ * the pre-PR bitmap implementations it replaces:
+ * <ul>
+ * <li>{@link Roaring64Bitmap} — previously used in {@code
InMemoryDelayedDeliveryTracker}.</li>
+ * <li>{@link RoaringBitmap} — previously used in {@code
ConsumerNameIndexTracker},
+ * {@code DrainingHashesTracker}, and the bucket delayed-delivery
family.</li>
+ * </ul>
+ *
+ * <p>Run with:
+ * <pre>{@code
+ * ./gradlew :microbench:shadowJar
+ * java -jar microbench/build/libs/microbench-*-benchmarks.jar
LongBitmapBenchmark
+ * }</pre>
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Benchmark)
+@Warmup(time = 2, iterations = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(time = 3, iterations = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(1)
+public class LongBitmapBenchmark {
+
+ @Param({"1000", "100000"})
+ public int bitmapSize;
+
+ private LongBitmap longBitmap;
+ private RoaringBitmap roaringBitmap;
+ private Roaring64Bitmap roaring64Bitmap;
+
+ private final AtomicLong nextValue = new AtomicLong();
+
+ @Setup(Level.Trial)
+ public void setup() {
+ longBitmap = LongBitmaps.create();
+ roaringBitmap = new RoaringBitmap();
+ roaring64Bitmap = new Roaring64Bitmap();
+ for (int i = 0; i < bitmapSize; i++) {
+ longBitmap.add(i);
+ roaringBitmap.add(i);
+ roaring64Bitmap.addLong(i);
+ }
+ nextValue.set(bitmapSize);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean longBitmapAddSingleThread() {
+ long v = nextValue.getAndIncrement();
+ return longBitmap.checkedAdd(v);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean roaringBitmapAddSingleThread() {
+ long v = nextValue.getAndIncrement();
+ return roaringBitmap.checkedAdd((int) v);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean roaring64BitmapAddSingleThread() {
+ long v = nextValue.getAndIncrement();
+ boolean existed = roaring64Bitmap.contains(v);
+ roaring64Bitmap.addLong(v);
+ return !existed;
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean longBitmapContainsSingleThread() {
+ return longBitmap.contains(nextValue.getAndIncrement() % bitmapSize);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean roaringBitmapContainsSingleThread() {
+ return roaringBitmap.contains((int) (nextValue.getAndIncrement() %
bitmapSize));
+ }
+
+ @Benchmark
+ @Threads(1)
+ public boolean roaring64BitmapContainsSingleThread() {
+ return roaring64Bitmap.contains(nextValue.getAndIncrement() %
bitmapSize);
+ }
+
+ // Bare RoaringBitmap variants are not thread-safe and are omitted below.
+
+ @Benchmark
+ @Threads(4)
+ public boolean longBitmapAdd4Threads() {
+ long v = nextValue.getAndIncrement();
+ return longBitmap.checkedAdd(v);
+ }
+
+ @Benchmark
+ @Threads(4)
+ public void longBitmapContains4Threads(Blackhole bh) {
+ int v = (int) (Thread.currentThread().getId() * 31 +
System.nanoTime());
+ bh.consume(longBitmap.contains(Math.abs(v) % bitmapSize));
+ }
+}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
index c2e1c63b400..b1e50700c72 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
@@ -36,7 +36,8 @@ import lombok.Getter;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import
org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
-import org.roaringbitmap.longlong.Roaring64Bitmap;
+import org.apache.pulsar.common.util.collections.LongBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmaps;
public class InMemoryDelayedDeliveryTracker extends
AbstractDelayedDeliveryTracker {
@@ -44,8 +45,8 @@ public class InMemoryDelayedDeliveryTracker extends
AbstractDelayedDeliveryTrack
protected final Logger log;
// timestamp -> ledgerId -> entryId
- // AVL tree -> OpenHashMap -> RoaringBitmap
- protected final Long2ObjectSortedMap<Long2ObjectSortedMap<Roaring64Bitmap>>
+ // AVL tree -> OpenHashMap -> LongBitmap
+ protected final Long2ObjectSortedMap<Long2ObjectSortedMap<LongBitmap>>
delayedMessageMap = new Long2ObjectAVLTreeMap<>();
// If we detect that all messages have fixed delay time, such that the
delivery is
@@ -141,14 +142,9 @@ public class InMemoryDelayedDeliveryTracker extends
AbstractDelayedDeliveryTrack
.log("Add message");
long timestamp = roundTimestamp(deliverAt);
- Roaring64Bitmap bitmap = delayedMessageMap.computeIfAbsent(timestamp,
k -> new Long2ObjectRBTreeMap<>())
- .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap());
- // Roaring64Bitmap does not store duplicates, so track if it a new
element
- // so we can keep delayedMessagesCount in sync
- boolean isNew = !bitmap.contains(entryId);
-
- if (isNew) {
- bitmap.addLong(entryId);
+ LongBitmap bitmap = delayedMessageMap.computeIfAbsent(timestamp, k ->
new Long2ObjectRBTreeMap<>())
+ .computeIfAbsent(ledgerId, k -> LongBitmaps.create());
+ if (bitmap.checkedAdd(entryId)) {
delayedMessagesCount.incrementAndGet();
}
@@ -221,28 +217,19 @@ public class InMemoryDelayedDeliveryTracker extends
AbstractDelayedDeliveryTrack
}
LongSet ledgerIdToDelete = new LongOpenHashSet();
- Long2ObjectSortedMap<Roaring64Bitmap> ledgerMap =
delayedMessageMap.get(timestamp);
- for (Long2ObjectMap.Entry<Roaring64Bitmap> ledgerEntry :
ledgerMap.long2ObjectEntrySet()) {
+ Long2ObjectSortedMap<LongBitmap> ledgerMap =
delayedMessageMap.get(timestamp);
+ for (Long2ObjectMap.Entry<LongBitmap> ledgerEntry :
ledgerMap.long2ObjectEntrySet()) {
long ledgerId = ledgerEntry.getLongKey();
- Roaring64Bitmap entryIds = ledgerEntry.getValue();
- long cardinality = entryIds.getLongCardinality();
- if (cardinality <= n) {
- int cardinalityInt = (int) cardinality;
- entryIds.forEach(entryId -> {
- positions.add(PositionFactory.create(ledgerId,
entryId));
- });
- n -= cardinalityInt;
- delayedMessagesCount.addAndGet(-cardinalityInt);
+ LongBitmap entryIds = ledgerEntry.getValue();
+ long cardinality = entryIds.cardinality();
+ long drained = entryIds.drainTo(n, entryId -> {
+ positions.add(PositionFactory.create(ledgerId, entryId));
+ });
+ delayedMessagesCount.addAndGet(-drained);
+ n -= drained;
+ if (drained == cardinality) {
+ // Bitmap is now empty; the entry will be removed from the
ledger map below.
ledgerIdToDelete.add(ledgerId);
- } else {
- Roaring64Bitmap entryIdsToRemove = new Roaring64Bitmap();
- entryIds.stream().limit(n).forEach(entryId -> {
- positions.add(PositionFactory.create(ledgerId,
entryId));
- entryIdsToRemove.addLong(entryId);
- });
- entryIds.andNot(entryIdsToRemove);
- delayedMessagesCount.addAndGet(-n);
- n = 0;
}
if (n <= 0) {
break;
@@ -286,16 +273,14 @@ public class InMemoryDelayedDeliveryTracker extends
AbstractDelayedDeliveryTrack
}
/**
- * This method rely on Roaring64Bitmap::getLongSizeInBytes to calculate
the memory usage of the buffer.
- * The memory usage of the buffer is not accurate, because
Roaring64Bitmap::getLongSizeInBytes will
- * overestimate the memory usage of the buffer a lot.
- * @return the memory usage of the buffer
+ * Estimates memory usage of all bitmaps in the tracker.
+ * Uses serialized size as an approximation of memory usage.
+ * @return estimated memory usage in bytes
*/
@Override
public long getBufferMemoryUsage() {
return delayedMessageMap.values().stream().mapToLong(
- ledgerMap -> ledgerMap.values().stream().mapToLong(
- Roaring64Bitmap::getLongSizeInBytes).sum()).sum();
+ ledgerMap ->
ledgerMap.values().stream().mapToLong(LongBitmap::serializedSize).sum()).sum();
}
@Override
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
index 2aad122752f..489478df562 100644
---
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
@@ -34,7 +34,8 @@ 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.roaringbitmap.RoaringBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmaps;
@CustomLog
@Data
@@ -55,7 +56,7 @@ abstract class Bucket {
long startLedgerId;
long endLedgerId;
- Map<Long, RoaringBitmap> delayedIndexBitMap;
+ Map<Long, LongBitmap> delayedIndexBitMap;
long numberBucketDelayedMessages;
@@ -76,7 +77,7 @@ abstract class Bucket {
}
boolean containsMessage(long ledgerId, long entryId) {
- RoaringBitmap bitSet = delayedIndexBitMap.get(ledgerId);
+ LongBitmap bitSet = delayedIndexBitMap.get(ledgerId);
if (bitSet == null) {
return false;
}
@@ -84,12 +85,12 @@ abstract class Bucket {
}
void putIndexBit(long ledgerId, long entryId) {
- delayedIndexBitMap.computeIfAbsent(ledgerId, k -> new
RoaringBitmap()).add(entryId, entryId + 1);
+ delayedIndexBitMap.computeIfAbsent(ledgerId, k ->
LongBitmaps.create()).add(entryId, entryId + 1);
}
boolean removeIndexBit(long ledgerId, long entryId) {
boolean contained = false;
- RoaringBitmap bitSet = delayedIndexBitMap.get(ledgerId);
+ LongBitmap bitSet = delayedIndexBitMap.get(ledgerId);
if (bitSet != null && bitSet.contains(entryId, entryId + 1)) {
contained = true;
bitSet.remove(entryId, entryId + 1);
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 9bfea0604d7..b8a8eb7f0f0 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
@@ -64,8 +64,8 @@ import org.apache.pulsar.broker.delayed.proto.SnapshotSegment;
import
org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
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.TripleLongPriorityQueue;
-import org.roaringbitmap.RoaringBitmap;
@ThreadSafe
public class BucketDelayedDeliveryTracker extends
AbstractDelayedDeliveryTracker {
@@ -562,7 +562,7 @@ public class BucketDelayedDeliveryTracker extends
AbstractDelayedDeliveryTracker
buckets.get(buckets.size() -
1).endLedgerId);
// Merge bit map to new bucket
- Map<Long, RoaringBitmap> delayedIndexBitMap =
+ Map<Long, LongBitmap> delayedIndexBitMap =
new
HashMap<>(buckets.get(0).getDelayedIndexBitMap());
for (int i = 1; i < buckets.size(); i++) {
buckets.get(i).delayedIndexBitMap.forEach((ledgerId, bitMapB) -> {
@@ -577,8 +577,6 @@ public class BucketDelayedDeliveryTracker extends
AbstractDelayedDeliveryTracker
});
}
- // optimize bm
-
delayedIndexBitMap.values().forEach(RoaringBitmap::runOptimize);
immutableBucketDelayedIndexPair.getLeft().setDelayedIndexBitMap(delayedIndexBitMap);
afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime);
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 e76b225b9d1..2124c6e9540 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
@@ -20,7 +20,8 @@ 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.NULL_LONG_PROMISE;
-import java.io.IOException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -36,8 +37,8 @@ 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.FutureUtil;
-import org.roaringbitmap.InvalidRoaringFormat;
-import org.roaringbitmap.RoaringBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmaps;
@CustomLog
class ImmutableBucket extends Bucket {
@@ -149,7 +150,6 @@ class ImmutableBucket extends Bucket {
/**
* Recover delayed index bit map and message numbers.
- * @throws InvalidRoaringFormat invalid bitmap serialization format
*/
private void recoverDelayedIndexBitMapAndNumber(int startSnapshotIndex,
SnapshotMetadata
snapshotMetadata) {
@@ -157,24 +157,22 @@ class ImmutableBucket extends Bucket {
final var numberMessages = new MutableLong(0);
for (int i = startSnapshotIndex; i <
snapshotMetadata.getMetadataListCount(); i++) {
snapshotMetadata.getMetadataAt(i).forEachDelayedIndexBitMap((ledgerId, bs) -> {
- final var sbm = new RoaringBitmap();
+ final ByteBuf buf = Unpooled.wrappedBuffer(bs);
try {
- sbm.deserialize(java.nio.ByteBuffer.wrap(bs));
- } catch (IOException e) {
- throw new InvalidRoaringFormat(e.getMessage());
+ final LongBitmap sbm = LongBitmaps.deserialize(buf);
+ numberMessages.add(sbm.cardinality());
+ delayedIndexBitMap.compute(ledgerId, (lId, bm) -> {
+ if (bm == null) {
+ return sbm;
+ }
+ bm.or(sbm);
+ return bm;
+ });
+ } finally {
+ buf.release();
}
- numberMessages.add(sbm.getCardinality());
- delayedIndexBitMap.compute(ledgerId, (lId, bm) -> {
- if (bm == null) {
- return sbm;
- }
- bm.or(sbm);
- return bm;
- });
});
}
- // optimize bm
- delayedIndexBitMap.values().forEach(RoaringBitmap::runOptimize);
setNumberBucketDelayedMessages(numberMessages.longValue());
}
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 75a3fad6589..a91016bba0b 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,7 +19,6 @@
package org.apache.pulsar.broker.delayed.bucket;
import static com.google.common.base.Preconditions.checkArgument;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
@@ -34,8 +33,9 @@ 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;
-import org.roaringbitmap.RoaringBitmap;
@CustomLog
class MutableBucket extends Bucket implements AutoCloseable {
@@ -73,9 +73,9 @@ class MutableBucket extends Bucket implements AutoCloseable {
List<SnapshotSegment> bucketSnapshotSegments = new ArrayList<>();
List<SnapshotSegmentMetadata> segmentMetadataList = new ArrayList<>();
- Map<Long, RoaringBitmap> immutableBucketBitMap = new HashMap<>();
+ Map<Long, LongBitmap> immutableBucketBitMap = new HashMap<>();
- Map<Long, RoaringBitmap> bitMap = new HashMap<>();
+ Map<Long, LongBitmap> bitMap = new HashMap<>();
SnapshotSegment snapshotSegment = new SnapshotSegment();
SnapshotSegmentMetadata segmentMetadata = new
SnapshotSegmentMetadata();
@@ -105,7 +105,7 @@ class MutableBucket extends Bucket implements AutoCloseable
{
sharedQueue.add(timestamp, ledgerId, entryId);
}
- bitMap.computeIfAbsent(ledgerId, k -> new
RoaringBitmap()).add(entryId, entryId + 1);
+ bitMap.computeIfAbsent(ledgerId, k ->
LongBitmaps.create()).add(entryId, entryId + 1);
numMessages++;
@@ -116,16 +116,12 @@ class MutableBucket extends Bucket implements
AutoCloseable {
segmentMetadata.setMinScheduleTimestamp(currentFirstTimestamp);
currentTimestampUpperLimit = 0;
- Iterator<Map.Entry<Long, RoaringBitmap>> iterator =
bitMap.entrySet().iterator();
+ Iterator<Map.Entry<Long, LongBitmap>> iterator =
bitMap.entrySet().iterator();
while (iterator.hasNext()) {
final var entry = iterator.next();
final var lId = entry.getKey();
final var bm = entry.getValue();
- bm.runOptimize();
- ByteBuffer byteBuffer =
ByteBuffer.allocate(bm.serializedSizeInBytes());
- bm.serialize(byteBuffer);
- byteBuffer.flip();
- segmentMetadata.putDelayedIndexBitMap(lId,
byteBuffer.array());
+ segmentMetadata.putDelayedIndexBitMap(lId, bm.serialize());
immutableBucketBitMap.compute(lId, (__, bm0) -> {
if (bm0 == null) {
return bm;
@@ -144,10 +140,6 @@ class MutableBucket extends Bucket implements
AutoCloseable {
}
}
- // optimize bm
- immutableBucketBitMap.values().forEach(RoaringBitmap::runOptimize);
- this.delayedIndexBitMap.values().forEach(RoaringBitmap::runOptimize);
-
SnapshotMetadata bucketSnapshotMetadata = new SnapshotMetadata();
for (SnapshotSegmentMetadata sm : segmentMetadataList) {
bucketSnapshotMetadata.addMetadata().copyFrom(sm);
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerNameIndexTracker.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerNameIndexTracker.java
index 1f93313ab1b..b6e5173f7e5 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerNameIndexTracker.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsumerNameIndexTracker.java
@@ -22,7 +22,8 @@ import java.util.HashMap;
import java.util.Map;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.commons.lang3.mutable.MutableInt;
-import org.roaringbitmap.RoaringBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmaps;
/**
* Tracks the used consumer name indexes for each consumer name.
@@ -56,18 +57,18 @@ class ConsumerNameIndexTracker {
}
/*
- * Tracks the used indexes for a consumer name using a RoaringBitmap.
+ * Tracks the used indexes for a consumer name using a LongBitmap.
* A specific index slot is used when the bit is set.
* When all bits are cleared, the customer name can be removed from
tracking.
*/
static class ConsumerNameIndexSlots {
- private RoaringBitmap indexSlots = new RoaringBitmap();
+ private LongBitmap indexSlots = LongBitmaps.create();
public int allocateIndexSlot() {
// find the first index that is not set, if there is no such
index, add a new one
int index = (int) indexSlots.nextAbsentValue(0);
if (index == -1) {
- index = indexSlots.getCardinality();
+ index = (int) indexSlots.cardinality();
}
indexSlots.add(index);
return index;
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/DrainingHashesTracker.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/DrainingHashesTracker.java
index 5393cf4d6a6..87ec1cf1470 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/DrainingHashesTracker.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/DrainingHashesTracker.java
@@ -24,7 +24,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.PrimitiveIterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -33,7 +32,8 @@ import lombok.ToString;
import org.apache.pulsar.common.policies.data.DrainingHash;
import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
import org.apache.pulsar.common.policies.data.stats.DrainingHashImpl;
-import org.roaringbitmap.RoaringBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmap;
+import org.apache.pulsar.common.util.collections.LongBitmaps;
/**
* A thread-safe map to store draining hashes in the consumer.
@@ -142,7 +142,7 @@ public class DrainingHashesTracker {
}
private class ConsumerDrainingHashesStats {
- private final RoaringBitmap drainingHashes = new RoaringBitmap();
+ private final LongBitmap drainingHashes = LongBitmaps.create();
private long drainingHashesClearedTotal;
private final ReentrantReadWriteLock statsLock = new
ReentrantReadWriteLock();
@@ -166,12 +166,8 @@ public class DrainingHashesTracker {
.attr("hash", hash)
.attr("empty", empty)
.attr("drainingHashesClearedTotal",
drainingHashesClearedTotal)
- .attr("cardinality", () ->
drainingHashes.getCardinality())
+ .attr("cardinality", () ->
drainingHashes.cardinality())
.log("Cleared hash in stats");
- if (empty) {
- // reduce memory usage by trimming the bitmap when the
RoaringBitmap instance is empty
- drainingHashes.trim();
- }
return empty;
} finally {
statsLock.writeLock().unlock();
@@ -181,11 +177,10 @@ public class DrainingHashesTracker {
public void updateConsumerStats(Consumer consumer, ConsumerStatsImpl
consumerStats) {
statsLock.readLock().lock();
try {
- int drainingHashesUnackedMessages = 0;
List<DrainingHash> drainingHashesStats = new ArrayList<>();
- PrimitiveIterator.OfInt hashIterator =
drainingHashes.stream().iterator();
- while (hashIterator.hasNext()) {
- int hash = hashIterator.nextInt();
+ int[] drainingHashesUnackedMessages = {0};
+ drainingHashes.forEachLong(hashLong -> {
+ int hash = (int) hashLong;
DrainingHashEntry entry = getEntry(hash);
if (entry == null) {
// Not-found entries are expected as a benign race
between the draining-hash
@@ -197,7 +192,7 @@ public class DrainingHashesTracker {
.attr("hash", hash)
.attr("consumer", consumer)
.log("Draining hash not found in the tracker
for consumer");
- continue;
+ return;
}
int unackedMessages = entry.getRefCount();
DrainingHashImpl drainingHash = new DrainingHashImpl();
@@ -205,11 +200,11 @@ public class DrainingHashesTracker {
drainingHash.unackMsgs = unackedMessages;
drainingHash.blockedAttempts = entry.getBlockedCount();
drainingHashesStats.add(drainingHash);
- drainingHashesUnackedMessages += unackedMessages;
- }
+ drainingHashesUnackedMessages[0] += unackedMessages;
+ });
consumerStats.drainingHashesCount = drainingHashesStats.size();
consumerStats.drainingHashesClearedTotal =
drainingHashesClearedTotal;
- consumerStats.drainingHashesUnackedMessages =
drainingHashesUnackedMessages;
+ consumerStats.drainingHashesUnackedMessages =
drainingHashesUnackedMessages[0];
consumerStats.drainingHashes = drainingHashesStats;
} finally {
statsLock.readLock().unlock();
diff --git a/pulsar-common/build.gradle.kts b/pulsar-common/build.gradle.kts
index ad7b37ca3f6..e7b7f28d579 100644
--- a/pulsar-common/build.gradle.kts
+++ b/pulsar-common/build.gradle.kts
@@ -159,6 +159,7 @@ dependencies {
api(libs.netty.handler)
api(libs.netty.buffer)
api(libs.netty.resolver.dns)
+ implementation(libs.roaringbitmap)
implementation(variantOf(libs.netty.transport.native.epoll) {
classifier("linux-x86_64") })
implementation(variantOf(libs.netty.transport.native.epoll) {
classifier("linux-aarch_64") })
implementation(libs.netty.transport.native.unix.common)
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentRoaringBitmap.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentRoaringBitmap.java
new file mode 100644
index 00000000000..7c65441004b
--- /dev/null
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentRoaringBitmap.java
@@ -0,0 +1,441 @@
+/*
+ * 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.common.util.collections;
+
+import io.netty.buffer.ByteBuf;
+import java.io.DataInput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.locks.StampedLock;
+import java.util.function.LongConsumer;
+import org.roaringbitmap.PeekableIntIterator;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+
+/**
+ * {@link LongBitmap} implementation backed by {@link MutableRoaringBitmap}
and guarded
+ * by a {@link StampedLock}.
+ *
+ * <p><b>Thread-safety basis.</b> RoaringBitmap is not thread-safe by default
+ * (see <a
href="https://github.com/apache/pulsar/issues/25991";>pulsar#25991</a>). This
+ * wrapper relies on the documented contract that {@link
MutableRoaringBitmap}'s read
+ * methods — the {@code ImmutableBitmapDataProvider} surface inherited from
+ * {@code ImmutableRoaringBitmap} — do not mutate internal state, while
methods added by
+ * {@code BitmapDataProvider} and other {@code MutableRoaringBitmap} mutators
+ * ({@code andNot}, {@code or}, {@code checkedRemove}, {@code runOptimize},
{@code clone},
+ * ...) do. Read methods run under the read lock; mutators under the write
lock.
+ * {@code clone()} is used under the read lock in {@link #forEachLong} and
{@link #serialize};
+ * its source has been audited to be read-only on the live bitmap. <b>Before
upgrading
+ * the RoaringBitmap dependency or changing the lock split</b>, re-audit these
methods
+ * and run the concurrency regression tests ({@code
testConcurrentForEachLongAndMutate},
+ * {@code testOrDoesNotMutateInput}).
+ *
+ * <p><b>Critical sections.</b> Single-value reads take the read lock;
mutations take the
+ * write lock. Bulk mutations that touch two bitmaps ({@link #or}) acquire
this bitmap's
+ * write lock and the other's read lock in {@code identityHashCode} order, so
concurrent
+ * {@code A.or(B)} and {@code B.or(A)} cannot deadlock. Long non-mutating work
+ * ({@link #serialize}, {@link #forEachLong}) clones under a brief read lock
and finishes
+ * without holding it, so optimize/iterate/runOptimize don't block writers.
+ *
+ * <p><b>Memory.</b> {@link MutableRoaringBitmap#trim()} fires when removals
since the
+ * last trim reach {@link #TRIM_AFTER_REMOVES}, or whenever the bitmap becomes
empty.
+ * {@link #serialize} runs {@code runOptimize()} on the clone so persisted
bytes are compact.
+ */
+class ConcurrentRoaringBitmap implements LongBitmap {
+
+ private static final long TRIM_AFTER_REMOVES = 10000;
+ private static final long UINT32_SIZE = 1L << 32;
+ private static final long MAX_UINT32 = UINT32_SIZE - 1;
+
+ private final MutableRoaringBitmap bitmap;
+ private final StampedLock lock;
+ private long removesSinceTrim;
+
+ ConcurrentRoaringBitmap() {
+ this.bitmap = new MutableRoaringBitmap();
+ this.lock = new StampedLock();
+ }
+
+ private ConcurrentRoaringBitmap(MutableRoaringBitmap bitmap) {
+ this.bitmap = bitmap;
+ this.lock = new StampedLock();
+ }
+
+ @Override
+ public void add(long value) {
+ validateRange(value);
+ long stamp = lock.writeLock();
+ try {
+ bitmap.add((int) value);
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public boolean checkedAdd(long value) {
+ validateRange(value);
+ long stamp = lock.writeLock();
+ try {
+ return bitmap.checkedAdd((int) value);
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public void add(long from, long to) {
+ if (to <= from) {
+ return;
+ }
+ validateRange(from);
+ validateRange(to - 1);
+ long stamp = lock.writeLock();
+ try {
+ bitmap.add(from, to);
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public void remove(long value) {
+ validateRange(value);
+ long stamp = lock.writeLock();
+ try {
+ if (bitmap.checkedRemove((int) value)) {
+ removesSinceTrim++;
+ maybeTrim();
+ }
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public void remove(long from, long to) {
+ if (to <= from) {
+ return;
+ }
+ validateRange(from);
+ validateRange(to - 1);
+ long stamp = lock.writeLock();
+ try {
+ bitmap.remove(from, to);
+ // Range size upper-bounds removals; clamp so a huge range can't
overflow the counter.
+ removesSinceTrim = Math.min(removesSinceTrim + (to - from),
TRIM_AFTER_REMOVES);
+ maybeTrim();
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public boolean contains(long value) {
+ if (value < 0 || value > MAX_UINT32) {
+ return false;
+ }
+ long stamp = lock.readLock();
+ try {
+ return bitmap.contains((int) value);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public boolean contains(long from, long to) {
+ if (from < 0 || from > MAX_UINT32 || to <= from) {
+ return false;
+ }
+ long stamp = lock.readLock();
+ try {
+ // Clamp: contains treats out-of-range `to` as a query past the
uint32 end, but
+ // RoaringBitmap.contains is unreliable when `to` exceeds
UINT32_SIZE.
+ return bitmap.contains(from, Math.min(to, UINT32_SIZE));
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public long cardinality() {
+ long stamp = lock.readLock();
+ try {
+ return bitmap.getLongCardinality();
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public boolean isEmpty() {
+ long stamp = lock.readLock();
+ try {
+ return bitmap.isEmpty();
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public long nextAbsentValue(long from) {
+ if (from < 0 || from > MAX_UINT32) {
+ return -1;
+ }
+ long stamp = lock.readLock();
+ try {
+ return bitmap.nextAbsentValue((int) from);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public void or(LongBitmap other) {
+ if (other == this) {
+ return;
+ }
+ if (!(other instanceof ConcurrentRoaringBitmap)) {
+ throw new IllegalArgumentException("Unsupported LongBitmap type: "
+ other.getClass());
+ }
+ ConcurrentRoaringBitmap that = (ConcurrentRoaringBitmap) other;
+
+ // Acquire this.writeLock + that.readLock in identityHashCode order so
concurrent
+ // A.or(B) and B.or(A) don't deadlock. Fall back to inner bitmap
identity on collision.
+ boolean thisFirst;
+ int outerCmp = Integer.compare(
+ System.identityHashCode(this), System.identityHashCode(that));
+ if (outerCmp != 0) {
+ thisFirst = outerCmp < 0;
+ } else {
+ thisFirst = System.identityHashCode(this.bitmap) <
System.identityHashCode(that.bitmap);
+ }
+
+ if (thisFirst) {
+ long thisStamp = this.lock.writeLock();
+ try {
+ long thatStamp = that.lock.readLock();
+ try {
+ this.bitmap.or(that.bitmap);
+ } finally {
+ that.lock.unlockRead(thatStamp);
+ }
+ } finally {
+ this.lock.unlockWrite(thisStamp);
+ }
+ } else {
+ long thatStamp = that.lock.readLock();
+ try {
+ long thisStamp = this.lock.writeLock();
+ try {
+ this.bitmap.or(that.bitmap);
+ } finally {
+ this.lock.unlockWrite(thisStamp);
+ }
+ } finally {
+ that.lock.unlockRead(thatStamp);
+ }
+ }
+ }
+
+ @Override
+ public void forEachLong(LongConsumer action) {
+ MutableRoaringBitmap snapshot;
+ long stamp = lock.readLock();
+ try {
+ snapshot = bitmap.clone();
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ snapshot.forEach((org.roaringbitmap.IntConsumer) v ->
+ action.accept(Integer.toUnsignedLong(v)));
+ }
+
+ @Override
+ public long drainTo(long limit, LongConsumer action) {
+ if (limit <= 0) {
+ return 0;
+ }
+ MutableRoaringBitmap toRemove = new MutableRoaringBitmap();
+ long collected;
+ long writeStamp = lock.writeLock();
+ try {
+ PeekableIntIterator it = bitmap.getIntIterator();
+ collected = 0;
+ while (collected < limit && it.hasNext()) {
+ toRemove.add(it.next());
+ collected++;
+ }
+ if (collected == 0) {
+ return 0;
+ }
+ bitmap.andNot(toRemove);
+ removesSinceTrim = Math.min(removesSinceTrim + collected,
TRIM_AFTER_REMOVES);
+ maybeTrim();
+ } finally {
+ lock.unlockWrite(writeStamp);
+ }
+
+ toRemove.forEach((org.roaringbitmap.IntConsumer) v ->
+ action.accept(Integer.toUnsignedLong(v)));
+ return collected;
+ }
+
+ @Override
+ public long serializedSize() {
+ long stamp = lock.readLock();
+ try {
+ return bitmap.serializedSizeInBytes();
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public byte[] serialize() {
+ MutableRoaringBitmap copy;
+ long stamp = lock.readLock();
+ try {
+ copy = bitmap.clone();
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ copy.runOptimize();
+ byte[] bytes = new byte[copy.serializedSizeInBytes()];
+ copy.serialize(ByteBuffer.wrap(bytes));
+ return bytes;
+ }
+
+ static ConcurrentRoaringBitmap deserialize(ByteBuf buf) {
+ try {
+ ByteBuffer nioBuffer = buf.nioBuffer(buf.readerIndex(),
buf.readableBytes());
+ int startPosition = nioBuffer.position();
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ bitmap.deserialize(new ByteBufferDataInput(nioBuffer));
+ buf.skipBytes(nioBuffer.position() - startPosition);
+ return new ConcurrentRoaringBitmap(bitmap);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to deserialize LongBitmap", e);
+ }
+ }
+
+ /**
+ * Trims the underlying bitmap if enough removals have accumulated or it's
empty.
+ * Caller must hold the write lock and have already updated {@link
#removesSinceTrim}.
+ */
+ private void maybeTrim() {
+ if (removesSinceTrim >= TRIM_AFTER_REMOVES || bitmap.isEmpty()) {
+ bitmap.trim();
+ removesSinceTrim = 0;
+ }
+ }
+
+ private static void validateRange(long value) {
+ if (value < 0 || value > MAX_UINT32) {
+ throw new IllegalArgumentException(
+ "Value out of range [0, " + MAX_UINT32 + "]: " + value);
+ }
+ }
+
+ /** Minimal {@link DataInput} over a {@link ByteBuffer} for RoaringBitmap
deserialization. */
+ private static final class ByteBufferDataInput implements DataInput {
+ private final ByteBuffer buffer;
+
+ ByteBufferDataInput(ByteBuffer buffer) {
+ this.buffer = buffer;
+ }
+
+ @Override
+ public void readFully(byte[] b) {
+ buffer.get(b);
+ }
+
+ @Override
+ public void readFully(byte[] b, int off, int len) {
+ buffer.get(b, off, len);
+ }
+
+ @Override
+ public int skipBytes(int n) {
+ int skip = Math.min(n, buffer.remaining());
+ buffer.position(buffer.position() + skip);
+ return skip;
+ }
+
+ @Override
+ public boolean readBoolean() {
+ return buffer.get() != 0;
+ }
+
+ @Override
+ public byte readByte() {
+ return buffer.get();
+ }
+
+ @Override
+ public int readUnsignedByte() {
+ return Byte.toUnsignedInt(buffer.get());
+ }
+
+ @Override
+ public short readShort() {
+ return buffer.getShort();
+ }
+
+ @Override
+ public int readUnsignedShort() {
+ return Short.toUnsignedInt(buffer.getShort());
+ }
+
+ @Override
+ public char readChar() {
+ return buffer.getChar();
+ }
+
+ @Override
+ public int readInt() {
+ return buffer.getInt();
+ }
+
+ @Override
+ public long readLong() {
+ return buffer.getLong();
+ }
+
+ @Override
+ public float readFloat() {
+ return buffer.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ return buffer.getDouble();
+ }
+
+ @Override
+ public String readLine() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String readUTF() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmap.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmap.java
new file mode 100644
index 00000000000..5176da8c125
--- /dev/null
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmap.java
@@ -0,0 +1,170 @@
+/*
+ * 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.common.util.collections;
+
+import java.util.function.LongConsumer;
+
+/**
+ * Thread-safe bitmap abstraction for tracking long values.
+ *
+ * <p>The current implementation supports values in the unsigned 32-bit range
+ * {@code [0, 2^32 - 1]}. Methods that modify the bitmap reject values outside
this
+ * range with {@link IllegalArgumentException}. Query methods return {@code
false}
+ * or {@code -1} for out-of-range values where applicable.
+ *
+ * <p>Supports point and range operations, bulk union, atomic draining,
iteration,
+ * and serialization. All operations are thread-safe.
+ *
+ * <p>This abstraction is used for high-throughput broker metadata tracking,
+ * including delayed-delivery tracking, consumer-name allocation, and
+ * draining-hash tracking.
+ *
+ * <p>Example:
+ * <pre>{@code
+ * LongBitmap bitmap = LongBitmaps.create();
+ * bitmap.add(12345L);
+ * if (bitmap.contains(12345L)) { ... }
+ *
+ * byte[] bytes = bitmap.serialize();
+ * LongBitmap restored =
LongBitmaps.deserialize(Unpooled.wrappedBuffer(bytes));
+ * }</pre>
+ */
+public interface LongBitmap {
+
+ /**
+ * Adds a value.
+ *
+ * @param value value to add, must be in {@code [0, 2^32 - 1]}
+ * @throws IllegalArgumentException if value is outside the supported range
+ */
+ void add(long value);
+
+ /**
+ * Adds a value if it is not already present.
+ *
+ * <p>This operation is atomic. Unlike {@code if (!contains(value))
add(value)},
+ * the check and add are performed as a single operation.
+ *
+ * @param value value to add, must be in {@code [0, 2^32 - 1]}
+ * @return {@code true} if the value was added, {@code false} if it
already existed
+ * @throws IllegalArgumentException if value is outside the supported range
+ */
+ boolean checkedAdd(long value);
+
+ /**
+ * Adds all values in the half-open range {@code [from, to)}.
+ *
+ * <p>No-op if {@code to <= from}.
+ *
+ * @param from inclusive lower bound
+ * @param to exclusive upper bound
+ * @throws IllegalArgumentException if the range exceeds the supported
value range
+ */
+ void add(long from, long to);
+
+ /**
+ * Removes a value. No-op if absent.
+ *
+ * @param value value to remove
+ * @throws IllegalArgumentException if value is outside the supported range
+ */
+ void remove(long value);
+
+ /**
+ * Removes all values in the half-open range {@code [from, to)}.
+ *
+ * <p>No-op if {@code to <= from}.
+ *
+ * @param from inclusive lower bound
+ * @param to exclusive upper bound
+ * @throws IllegalArgumentException if the range exceeds the supported
value range
+ */
+ void remove(long from, long to);
+
+ /**
+ * Returns whether the bitmap contains the given value.
+ *
+ * @param value value to check
+ * @return {@code true} if present, otherwise {@code false}
+ */
+ boolean contains(long value);
+
+ /**
+ * Returns whether all values in {@code [from, to)} are present.
+ *
+ * @param from inclusive lower bound
+ * @param to exclusive upper bound
+ * @return {@code true} if all values in the range are present
+ */
+ boolean contains(long from, long to);
+
+ /** Returns the number of values currently stored. */
+ long cardinality();
+
+ /** Returns {@code true} if no values are stored. */
+ boolean isEmpty();
+
+ /**
+ * Returns the smallest absent value greater than or equal to {@code from}.
+ *
+ * @param from inclusive lower bound
+ * @return next absent value, or {@code -1} if none exists
+ */
+ long nextAbsentValue(long from);
+
+ /**
+ * Adds all values from {@code other} into this bitmap.
+ *
+ * @param other bitmap to merge
+ */
+ void or(LongBitmap other);
+
+ /**
+ * Iterates values in ascending order.
+ *
+ * <p>The iteration observes a stable view of the bitmap. Implementations
may
+ * choose the mechanism used to provide this guarantee.
+ *
+ * @param action callback invoked for each value
+ */
+ void forEachLong(LongConsumer action);
+
+ /**
+ * Atomically removes up to {@code limit} values and invokes {@code action}
+ * for each removed value.
+ *
+ * <p>Selection and removal are performed atomically. The callback is
invoked
+ * after removal has completed.
+ *
+ * @param limit maximum number of values to drain
+ * @param action callback invoked for each removed value
+ * @return number of values drained
+ */
+ long drainTo(long limit, LongConsumer action);
+
+ /**
+ * Returns an upper bound of the serialized size.
+ */
+ long serializedSize();
+
+ /**
+ * Serializes the bitmap into a newly allocated byte array.
+ */
+ byte[] serialize();
+}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmaps.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmaps.java
new file mode 100644
index 00000000000..489502a439d
--- /dev/null
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/LongBitmaps.java
@@ -0,0 +1,61 @@
+/*
+ * 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.common.util.collections;
+
+import io.netty.buffer.ByteBuf;
+
+/**
+ * Factory for creating {@link LongBitmap} instances.
+ */
+public final class LongBitmaps {
+
+ private LongBitmaps() {
+ // utility class
+ }
+
+ /**
+ * Creates a new empty thread-safe LongBitmap.
+ *
+ * @return a new LongBitmap instance
+ */
+ public static LongBitmap create() {
+ return new ConcurrentRoaringBitmap();
+ }
+
+ /**
+ * Deserializes a LongBitmap from a ByteBuf.
+ *
+ * <p>Advances the buffer's {@code readerIndex} by the number of bytes
consumed. The
+ * buffer may be heap-backed, direct, or a {@link
io.netty.buffer.CompositeByteBuf} —
+ * the implementation reads via {@link ByteBuf#nioBuffer(int, int)}
without copying
+ * when possible.
+ *
+ * <p>The serialized format is the standard 32-bit RoaringBitmap portable
format, so
+ * buffers produced by {@link LongBitmap#serialize()} round-trip exactly.
Buffers in
+ * other formats (e.g. {@code Roaring64Bitmap}) are rejected.
+ *
+ * @param buf the input buffer positioned at the start of the serialized
bitmap
+ * @return the deserialized LongBitmap
+ * @throws RuntimeException if the buffer is malformed, truncated, or in an
+ * unrecognized format (wraps the underlying {@link
java.io.IOException})
+ */
+ public static LongBitmap deserialize(ByteBuf buf) {
+ return ConcurrentRoaringBitmap.deserialize(buf);
+ }
+}
diff --git
a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapCompatibilityTest.java
b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapCompatibilityTest.java
new file mode 100644
index 00000000000..6c9081db44f
--- /dev/null
+++
b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapCompatibilityTest.java
@@ -0,0 +1,235 @@
+/*
+ * 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.common.util.collections;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import org.roaringbitmap.RoaringBitmap;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+import org.testng.annotations.Test;
+
+/**
+ * Verifies the serialization compatibility characteristics of {@link
LongBitmap}.
+ *
+ * <p>Context: the previous implementation used two different RoaringBitmap
variants:
+ * <ul>
+ * <li>{@code InMemoryDelayedDeliveryTracker} used {@link Roaring64Bitmap}
(in-memory only)
+ * <li>{@code BucketDelayedDeliveryTracker} used {@link RoaringBitmap}
(32-bit, persisted)
+ * </ul>
+ *
+ * <p>The new {@link LongBitmap} abstraction always uses the 32-bit {@link
RoaringBitmap}
+ * internally. These tests document and verify:
+ * <ol>
+ * <li>32-bit buffer compatibility: {@link LongBitmap} buffers are
byte-identical to standard
+ * {@link RoaringBitmap} buffers (required for {@code
BucketDelayedDeliveryTracker}
+ * backward compatibility — old persisted snapshots must round-trip).
+ * <li>64-bit buffer incompatibility: {@link Roaring64Bitmap} buffers cannot
be deserialized
+ * by {@link LongBitmap} (format includes high-32-bit bucket prefixes).
Migration from
+ * {@code Roaring64Bitmap} is safe only for in-memory state; persisted
state would require
+ * a one-time entry-by-entry rebuild.
+ * <li>Behavioral equivalence within {@code uint32} range: {@link
LongBitmap} and
+ * {@link Roaring64Bitmap} produce identical add/remove/contains results
for all
+ * values in {@code [0, 2^32)}.
+ * </ol>
+ */
+public class LongBitmapCompatibilityTest {
+
+ /**
+ * LongBitmap's serialized bytes must be identical to a 32-bit
RoaringBitmap's
+ * for the same uint32 values. This guarantees persisted snapshots written
by
+ * the old {@code BucketDelayedDeliveryTracker} (using RoaringBitmap)
remain
+ * readable, and vice versa.
+ */
+ @Test
+ public void testLongBitmapBufferEqualsStandardRoaringBitmap() throws
Exception {
+ LongBitmap longBitmap = LongBitmaps.create();
+ longBitmap.add(0);
+ longBitmap.add(100);
+ longBitmap.add(1L << 20);
+ longBitmap.add(0xFFFFFFFFL); // uint32 max
+
+ RoaringBitmap roaring = new RoaringBitmap();
+ roaring.add(0);
+ roaring.add(100);
+ roaring.add(1 << 20);
+ roaring.add(0xFFFFFFFF); // -1 as signed int
+
+ byte[] longBitmapBytes = serializeLongBitmap(longBitmap);
+ byte[] roaringBytes = serializeRoaring32(roaring);
+
+ assertEquals(longBitmapBytes, roaringBytes,
+ "LongBitmap buffer must be byte-identical to standard 32-bit
RoaringBitmap buffer");
+ }
+
+ /**
+ * Round-trip: 32-bit RoaringBitmap buffer -> LongBitmap. Confirms that
+ * persisted data written by the old code path can be read by LongBitmap.
+ */
+ @Test
+ public void testDeserializeFromRoaring32Buffer() throws Exception {
+ RoaringBitmap roaring = new RoaringBitmap();
+ for (int i = 0; i < 1000; i += 7) {
+ roaring.add(i);
+ }
+ roaring.add(0xFFFFFFFF);
+
+ byte[] roaringBytes = serializeRoaring32(roaring);
+ ByteBuf buf = Unpooled.wrappedBuffer(roaringBytes);
+ try {
+ LongBitmap longBitmap = LongBitmaps.deserialize(buf);
+ assertEquals(longBitmap.cardinality(),
roaring.getLongCardinality());
+ assertTrue(longBitmap.contains(0xFFFFFFFFL));
+ for (int i = 0; i < 1000; i += 7) {
+ assertTrue(longBitmap.contains(i), "missing " + i);
+ }
+ for (int i = 1; i < 1000; i += 7) {
+ assertFalse(longBitmap.contains(i));
+ }
+ } finally {
+ buf.release();
+ }
+ }
+
+ /**
+ * LongBitmap buffer cannot be deserialized as a Roaring64Bitmap.
+ * The 32-bit RoaringBitmap portable format (cookie 12346, ~22 bytes for
small sets)
+ * is shorter than Roaring64Bitmap's bucketed header, so {@code
Roaring64Bitmap.deserialize}
+ * throws {@link java.nio.BufferUnderflowException}.
+ *
+ * <p>Migration implication: persisted 32-bit data cannot be read by old
code paths
+ * that still expect Roaring64Bitmap, and vice versa.
+ */
+ @Test
+ public void testLongBitmapBufferNotReadableAsRoaring64() throws Exception {
+ LongBitmap longBitmap = LongBitmaps.create();
+ longBitmap.add(1);
+ longBitmap.add(100);
+ longBitmap.add(1000);
+
+ byte[] longBitmapBytes = serializeLongBitmap(longBitmap);
+
+ Roaring64Bitmap roaring64 = new Roaring64Bitmap();
+ // 32-bit format is shorter than 64-bit header expects — buffer
underflow or related I/O error.
+ assertThrows(Exception.class,
+ () -> roaring64.deserialize(ByteBuffer.wrap(longBitmapBytes)));
+ }
+
+ /**
+ * Roaring64Bitmap buffer cannot be deserialized as a LongBitmap.
+ * Roaring64Bitmap's serialized format starts with a bucket count, not the
+ * 32-bit cookie (12346), so {@code MutableRoaringBitmap.deserialize}
throws.
+ */
+ @Test
+ public void testRoaring64BufferNotReadableAsLongBitmap() throws Exception {
+ Roaring64Bitmap roaring64 = new Roaring64Bitmap();
+ roaring64.addLong(1);
+ roaring64.addLong(100);
+ roaring64.addLong(1000);
+
+ byte[] roaring64Bytes = serializeRoaring64(roaring64);
+
+ ByteBuf buf = Unpooled.wrappedBuffer(roaring64Bytes);
+ try {
+ // MutableRoaringBitmap wraps IOException as RuntimeException.
+ assertThrows(Exception.class, () -> LongBitmaps.deserialize(buf));
+ } finally {
+ buf.release();
+ }
+ }
+
+ /**
+ * Behavioral equivalence within uint32 range: LongBitmap and
Roaring64Bitmap
+ * produce identical results for all operations on values in [0, 2^32).
+ * This is what makes the InMemoryDelayedDeliveryTracker migration safe —
+ * BookKeeper entry IDs are currently always < 2^32.
+ */
+ @Test
+ public void testBehavioralEquivalenceWithinUint32() {
+ LongBitmap longBitmap = LongBitmaps.create();
+ Roaring64Bitmap roaring64 = new Roaring64Bitmap();
+
+ long[] values = {0, 1, 100, 65535, 65536, 1L << 20, 1L << 30,
0xFFFFFFFFL};
+ for (long v : values) {
+ longBitmap.add(v);
+ roaring64.addLong(v);
+ }
+
+ assertEquals(longBitmap.cardinality(), roaring64.getLongCardinality());
+ for (long v : values) {
+ assertTrue(longBitmap.contains(v));
+ assertTrue(roaring64.contains(v));
+ }
+
+ long[] toRemove = {0, 100, 65536, 1L << 30};
+ for (long v : toRemove) {
+ longBitmap.remove(v);
+ roaring64.removeLong(v);
+ }
+
+ assertEquals(longBitmap.cardinality(), roaring64.getLongCardinality());
+ for (long v : toRemove) {
+ assertFalse(longBitmap.contains(v));
+ assertFalse(roaring64.contains(v));
+ }
+ }
+
+ /**
+ * LongBitmap accepts the uint32 boundary (2^32 - 1) but rejects 2^32 and
above.
+ * Migration from Roaring64Bitmap must ensure no values >= 2^32 are
present;
+ * otherwise the migration would silently drop or reject those entries.
+ */
+ @Test
+ public void testUint32Boundary() {
+ LongBitmap longBitmap = LongBitmaps.create();
+ longBitmap.add(0xFFFFFFFFL);
+ assertTrue(longBitmap.contains(0xFFFFFFFFL));
+ assertEquals(longBitmap.cardinality(), 1);
+
+ assertThrows(IllegalArgumentException.class, () ->
longBitmap.add(0x100000000L));
+ assertThrows(IllegalArgumentException.class, () -> longBitmap.add(-1));
+ }
+
+ private static byte[] serializeLongBitmap(LongBitmap bitmap) {
+ return bitmap.serialize();
+ }
+
+ private static byte[] serializeRoaring32(RoaringBitmap bitmap) throws
Exception {
+ bitmap.runOptimize();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos);
+ bitmap.serialize(dos);
+ dos.close();
+ return baos.toByteArray();
+ }
+
+ private static byte[] serializeRoaring64(Roaring64Bitmap bitmap) throws
Exception {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos);
+ bitmap.serialize(dos);
+ dos.close();
+ return baos.toByteArray();
+ }
+}
diff --git
a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapTest.java
b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapTest.java
new file mode 100644
index 00000000000..35bbf9b5f57
--- /dev/null
+++
b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/LongBitmapTest.java
@@ -0,0 +1,824 @@
+/*
+ * 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.common.util.collections;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.testng.annotations.Test;
+
+/**
+ * Unit tests for {@link LongBitmap}. Covers point/range operations, uint32
boundary
+ * behavior, serialization round-trip, drain-to atomicity, and concurrency
contracts
+ * (or lock ordering, snapshot safety under mutation, or-input immutability).
+ */
+public class LongBitmapTest {
+
+ @Test
+ public void testBasicOperations() {
+ LongBitmap bitmap = LongBitmaps.create();
+
+ assertEquals(bitmap.cardinality(), 0);
+ assertFalse(bitmap.contains(1));
+
+ bitmap.add(1);
+ bitmap.add(100);
+ bitmap.add(1000);
+
+ assertEquals(bitmap.cardinality(), 3);
+ assertTrue(bitmap.contains(1));
+ assertTrue(bitmap.contains(100));
+ assertTrue(bitmap.contains(1000));
+ assertFalse(bitmap.contains(2));
+
+ bitmap.remove(100);
+ assertEquals(bitmap.cardinality(), 2);
+ assertFalse(bitmap.contains(100));
+ }
+
+ @Test
+ public void testRangeValidation() {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(0);
+ bitmap.add(0xFFFFFFFFL);
+ assertTrue(bitmap.contains(0));
+ assertTrue(bitmap.contains(0xFFFFFFFFL));
+
+ assertThrows(IllegalArgumentException.class, () -> bitmap.add(-1));
+ assertThrows(IllegalArgumentException.class, () ->
bitmap.add(0x100000000L));
+ assertFalse(bitmap.contains(-1));
+ assertFalse(bitmap.contains(0x100000000L));
+ }
+
+ @Test
+ public void testSerialization() {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 1000; i += 10) {
+ bitmap.add(i);
+ }
+
+ byte[] bytes = bitmap.serialize();
+ assertTrue(bytes.length > 0);
+
+ LongBitmap deserialized =
LongBitmaps.deserialize(Unpooled.wrappedBuffer(bytes));
+ assertEquals(deserialized.cardinality(), 100);
+ for (int i = 0; i < 1000; i += 10) {
+ assertTrue(deserialized.contains(i));
+ }
+ assertFalse(deserialized.contains(5));
+ }
+
+ @Test
+ public void testDeserializeFromVariousByteBufTypes() {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 2000; i += 7) {
+ bitmap.add(i);
+ }
+ byte[] bytes = bitmap.serialize();
+
+ ByteBuf heap = Unpooled.wrappedBuffer(bytes);
+ try {
+ LongBitmap r = LongBitmaps.deserialize(heap);
+ assertEquals(r.cardinality(), bitmap.cardinality());
+ assertEquals(heap.readerIndex(), bytes.length);
+ } finally {
+ heap.release();
+ }
+
+ ByteBuf direct = Unpooled.directBuffer(bytes.length);
+ try {
+ direct.writeBytes(bytes);
+ LongBitmap r = LongBitmaps.deserialize(direct);
+ assertEquals(r.cardinality(), bitmap.cardinality());
+ assertEquals(direct.readerIndex(), bytes.length);
+ } finally {
+ direct.release();
+ }
+
+ ByteBuf singleComp = Unpooled.wrappedBuffer(new
ByteBuf[]{Unpooled.wrappedBuffer(bytes)});
+ try {
+ LongBitmap r = LongBitmaps.deserialize(singleComp);
+ assertEquals(r.cardinality(), bitmap.cardinality());
+ assertEquals(singleComp.readerIndex(), bytes.length);
+ } finally {
+ singleComp.release();
+ }
+
+ int split = bytes.length / 2;
+ ByteBuf part1 = Unpooled.wrappedBuffer(bytes, 0, split);
+ ByteBuf part2 = Unpooled.wrappedBuffer(bytes, split, bytes.length -
split);
+ ByteBuf composite = Unpooled.wrappedBuffer(part1, part2);
+ try {
+ LongBitmap r = LongBitmaps.deserialize(composite);
+ assertEquals(r.cardinality(), bitmap.cardinality());
+ assertEquals(composite.readerIndex(), bytes.length);
+ } finally {
+ composite.release();
+ }
+ }
+
+ @Test
+ public void testAddIsIdempotent() {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(42);
+ assertEquals(bitmap.cardinality(), 1);
+ assertTrue(bitmap.contains(42));
+
+ bitmap.add(42);
+ bitmap.add(42);
+ assertEquals(bitmap.cardinality(), 1);
+ assertTrue(bitmap.contains(42));
+ }
+
+ @Test
+ public void testCheckedAdd() {
+ LongBitmap bitmap = LongBitmaps.create();
+
+ assertTrue(bitmap.checkedAdd(42));
+ assertTrue(bitmap.contains(42));
+ assertEquals(bitmap.cardinality(), 1);
+
+ assertFalse(bitmap.checkedAdd(42));
+ assertEquals(bitmap.cardinality(), 1);
+
+ assertTrue(bitmap.checkedAdd(100));
+ assertEquals(bitmap.cardinality(), 2);
+
+ assertThrows(IllegalArgumentException.class, () ->
bitmap.checkedAdd(-1));
+ assertThrows(IllegalArgumentException.class, () ->
bitmap.checkedAdd(0x100000000L));
+ assertEquals(bitmap.cardinality(), 2);
+ }
+
+ @Test
+ public void testForEach() {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(1);
+ bitmap.add(5);
+ bitmap.add(10);
+
+ List<Long> values = new ArrayList<>();
+ bitmap.forEachLong(values::add);
+
+ assertEquals(values.size(), 3);
+ Collections.sort(values);
+ assertEquals(values.get(0).longValue(), 1);
+ assertEquals(values.get(1).longValue(), 5);
+ assertEquals(values.get(2).longValue(), 10);
+ }
+
+ @Test
+ public void testConcurrentReads() throws Exception {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 10000; i++) {
+ bitmap.add(i);
+ }
+
+ int numThreads = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch latch = new CountDownLatch(numThreads);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ for (int i = 0; i < numThreads; i++) {
+ executor.submit(() -> {
+ try {
+ for (int j = 0; j < 1000; j++) {
+ if (!bitmap.contains(j)) {
+ errors.incrementAndGet();
+ }
+ if (bitmap.cardinality() != 10000) {
+ errors.incrementAndGet();
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+ assertEquals(errors.get(), 0);
+ }
+
+ @Test
+ public void testConcurrentWrites() throws Exception {
+ LongBitmap bitmap = LongBitmaps.create();
+
+ int numThreads = 10;
+ int valuesPerThread = 1000;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch latch = new CountDownLatch(numThreads);
+
+ for (int t = 0; t < numThreads; t++) {
+ int threadId = t;
+ executor.submit(() -> {
+ try {
+ for (int i = 0; i < valuesPerThread; i++) {
+ bitmap.add(threadId * valuesPerThread + i);
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+
+ assertEquals(bitmap.cardinality(), numThreads * valuesPerThread);
+ for (int t = 0; t < numThreads; t++) {
+ for (int i = 0; i < valuesPerThread; i++) {
+ assertTrue(bitmap.contains(t * valuesPerThread + i));
+ }
+ }
+ }
+
+ @Test
+ public void testConcurrentReadWrite() throws Exception {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 5000; i++) {
+ bitmap.add(i);
+ }
+
+ int numReaders = 5;
+ int numWriters = 5;
+ ExecutorService executor = Executors.newFixedThreadPool(numReaders +
numWriters);
+ CountDownLatch latch = new CountDownLatch(numReaders + numWriters);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ for (int i = 0; i < numReaders; i++) {
+ executor.submit(() -> {
+ try {
+ for (int j = 0; j < 1000; j++) {
+ bitmap.contains(j % 5000);
+ bitmap.cardinality();
+ }
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+ for (int i = 0; i < numWriters; i++) {
+ int writerId = i;
+ executor.submit(() -> {
+ try {
+ for (int j = 0; j < 1000; j++) {
+ bitmap.add(5000 + writerId * 1000 + j);
+ }
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+ assertEquals(errors.get(), 0);
+ }
+
+ @Test
+ public void testConcurrentForEachLongAndMutate() throws Exception {
+ // forEachLong takes a clone() snapshot under read lock; concurrent
mutations on the
+ // live bitmap must never corrupt the snapshot. Regression guard for
pulsar#25991.
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 10000; i++) {
+ bitmap.add(i);
+ }
+
+ int numReaders = 5;
+ int numWriters = 5;
+ ExecutorService executor = Executors.newFixedThreadPool(numReaders +
numWriters);
+ CountDownLatch latch = new CountDownLatch(numReaders + numWriters);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ for (int i = 0; i < numReaders; i++) {
+ executor.submit(() -> {
+ try {
+ for (int j = 0; j < 100; j++) {
+ long[] last = {-1};
+ bitmap.forEachLong(v -> {
+ // Snapshot values must arrive in ascending order.
+ if (v <= last[0]) {
+ errors.incrementAndGet();
+ }
+ last[0] = v;
+ });
+ }
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ for (int i = 0; i < numWriters; i++) {
+ int id = i;
+ executor.submit(() -> {
+ try {
+ for (int j = 0; j < 1000; j++) {
+ long v = 10000 + id * 1000 + (j % 1000);
+ bitmap.add(v);
+ bitmap.remove(v);
+ }
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+ assertEquals(errors.get(), 0);
+ assertEquals(bitmap.cardinality(), 10000);
+ }
+
+ @Test
+ public void testMemoryTrim() {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 100000; i++) {
+ bitmap.add(i);
+ }
+ for (int i = 0; i < 50000; i++) {
+ bitmap.remove(i);
+ }
+
+ assertEquals(bitmap.cardinality(), 50000);
+ for (int i = 0; i < 50000; i++) {
+ assertFalse(bitmap.contains(i));
+ }
+ for (int i = 50000; i < 100000; i++) {
+ assertTrue(bitmap.contains(i));
+ }
+ }
+
+ @Test
+ public void testIsEmpty() {
+ LongBitmap bitmap = LongBitmaps.create();
+ assertTrue(bitmap.isEmpty());
+
+ bitmap.add(1);
+ assertFalse(bitmap.isEmpty());
+
+ bitmap.remove(1);
+ assertTrue(bitmap.isEmpty());
+
+ bitmap.remove(999); // never added
+ assertTrue(bitmap.isEmpty());
+ }
+
+ @Test
+ public void testOr() {
+ LongBitmap a = LongBitmaps.create();
+ a.add(1);
+ a.add(100);
+ a.add(1000);
+
+ LongBitmap b = LongBitmaps.create();
+ b.add(100); // overlap
+ b.add(2000); // unique to b
+
+ a.or(b);
+
+ assertEquals(a.cardinality(), 4);
+ assertTrue(a.contains(1));
+ assertTrue(a.contains(100));
+ assertTrue(a.contains(1000));
+ assertTrue(a.contains(2000));
+
+ // b is not modified
+ assertEquals(b.cardinality(), 2);
+ }
+
+ @Test
+ public void testOrEmpty() {
+ LongBitmap a = LongBitmaps.create();
+ a.add(1);
+ a.add(2);
+
+ LongBitmap empty = LongBitmaps.create();
+ a.or(empty);
+ assertEquals(a.cardinality(), 2);
+
+ LongBitmap target = LongBitmaps.create();
+ target.or(a);
+ assertEquals(target.cardinality(), 2);
+ }
+
+ @Test
+ public void testOrSelfIsNoOp() {
+ LongBitmap a = LongBitmaps.create();
+ a.add(1);
+ a.add(100);
+
+ a.or(a); // should not deadlock
+
+ assertEquals(a.cardinality(), 2);
+ assertTrue(a.contains(1));
+ assertTrue(a.contains(100));
+ }
+
+ @Test
+ public void testRangeAddRemoveContains() {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(100, 200);
+ assertEquals(bitmap.cardinality(), 100);
+ for (long v = 100; v < 200; v++) {
+ assertTrue(bitmap.contains(v));
+ }
+ assertFalse(bitmap.contains(99));
+ assertFalse(bitmap.contains(200));
+
+ // Single-value range [x, x+1) is equivalent to add(x).
+ LongBitmap single = LongBitmaps.create();
+ single.add(42, 43);
+ assertTrue(single.contains(42, 43));
+ assertTrue(single.contains(42));
+ assertEquals(single.cardinality(), 1);
+
+ // Range contains: true iff EVERY value in [from, to) is set.
+ assertTrue(bitmap.contains(100, 200));
+ assertTrue(bitmap.contains(150, 160));
+ assertFalse(bitmap.contains(99, 101));
+ assertFalse(bitmap.contains(199, 201));
+ assertFalse(bitmap.contains(200, 300));
+
+ bitmap.remove(100, 150);
+ assertEquals(bitmap.cardinality(), 50);
+ for (long v = 100; v < 150; v++) {
+ assertFalse(bitmap.contains(v));
+ }
+ for (long v = 150; v < 200; v++) {
+ assertTrue(bitmap.contains(v));
+ }
+ }
+
+ @Test
+ public void testRangeVsSingleValueEquivalence() {
+ // For any v, add(v, v+1) / contains(v, v+1) / remove(v, v+1) ≡ add(v)
/ contains(v) / remove(v).
+ long[] values = {0, 1, 100, 65535, 65536, 1L << 30, 0xFFFFFFFFL};
+ for (long v : values) {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(v);
+ assertTrue(bitmap.contains(v));
+ assertEquals(bitmap.cardinality(), 1);
+
+ bitmap.add(v); // idempotent
+ assertEquals(bitmap.cardinality(), 1);
+
+ bitmap.remove(v);
+ assertFalse(bitmap.contains(v));
+ assertEquals(bitmap.cardinality(), 0);
+ }
+ }
+
+ @Test
+ public void testNextAbsentValue() {
+ LongBitmap bitmap = LongBitmaps.create();
+ // Empty bitmap: first absent is 0
+ assertEquals(bitmap.nextAbsentValue(0), 0);
+
+ bitmap.add(0);
+ bitmap.add(1);
+ bitmap.add(2);
+ // [0,1,2] present, next absent from 0 is 3
+ assertEquals(bitmap.nextAbsentValue(0), 3);
+
+ bitmap.add(5);
+ // Gap at 3,4
+ assertEquals(bitmap.nextAbsentValue(0), 3);
+ assertEquals(bitmap.nextAbsentValue(3), 3);
+ assertEquals(bitmap.nextAbsentValue(5), 6);
+
+ // Out of range returns -1
+ assertEquals(bitmap.nextAbsentValue(-1), -1);
+ assertEquals(bitmap.nextAbsentValue(0x100000000L), -1);
+ }
+
+ @Test
+ public void testDrainTo() {
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 100; i++) {
+ bitmap.add(i * 10); // 0, 10, 20, ..., 990
+ }
+
+ List<Long> drained = new ArrayList<>();
+ long count = bitmap.drainTo(5, drained::add);
+
+ assertEquals(count, 5);
+ assertEquals(drained.size(), 5);
+ assertEquals(drained.get(0).longValue(), 0);
+ assertEquals(drained.get(4).longValue(), 40);
+ assertEquals(bitmap.cardinality(), 95);
+ assertFalse(bitmap.contains(0));
+ assertFalse(bitmap.contains(40));
+ assertTrue(bitmap.contains(50));
+
+ drained.clear();
+ count = bitmap.drainTo(1000, drained::add); // drain all remaining
+ assertEquals(count, 95);
+ assertTrue(bitmap.isEmpty());
+ }
+
+ @Test
+ public void testDrainToZeroOrNegative() {
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(1);
+ bitmap.add(2);
+
+ assertEquals(bitmap.drainTo(0, v -> {
+ }), 0);
+ assertEquals(bitmap.cardinality(), 2);
+
+ assertEquals(bitmap.drainTo(-1, v -> {
+ }), 0);
+ assertEquals(bitmap.cardinality(), 2);
+ }
+
+ @Test
+ public void testEmptyRangeIsNoOp() {
+ // add(x, x) and remove(x, x) must be no-ops, not throw.
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(1);
+ bitmap.add(2);
+
+ bitmap.add(5, 5);
+ bitmap.add(10, 1); // reversed range
+ assertEquals(bitmap.cardinality(), 2);
+
+ bitmap.remove(5, 5);
+ bitmap.remove(10, 1);
+ assertEquals(bitmap.cardinality(), 2);
+ }
+
+ @Test
+ public void testDrainToFollowedByDrainToDoesNotLeak() {
+ // drainTo must respect the same trim threshold as remove(value).
+ // Add many values, drain in small batches; if trim never fires, the
bitmap
+ // accumulates unused container capacity across calls.
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 50000; i++) {
+ bitmap.add(i);
+ }
+
+ // Drain 100 at a time — well under TRIM_AFTER_REMOVES (10000) per
call,
+ // but cumulative trim counter should cross the threshold after
several batches.
+ long totalDrained = 0;
+ while (!bitmap.isEmpty()) {
+ totalDrained += bitmap.drainTo(100, v -> {
+ });
+ }
+ assertEquals(totalDrained, 50000);
+ assertTrue(bitmap.isEmpty());
+ }
+
+ @Test
+ public void testOrDoesNotMutateInput() {
+ // A.or(B) must treat B as read-only — required for our lock split
+ // (this=writeLock, other=readLock).
+ LongBitmap a = LongBitmaps.create();
+ LongBitmap b = LongBitmaps.create();
+ for (int i = 0; i < 1000; i++) {
+ a.add(i * 2);
+ b.add(i * 2 + 1);
+ }
+ long bCardinalityBefore = b.cardinality();
+
+ a.or(b);
+
+ assertEquals(b.cardinality(), bCardinalityBefore);
+ for (int i = 0; i < 1000; i++) {
+ assertTrue(b.contains(i * 2 + 1), "b lost value after a.or(b)");
+ assertFalse(b.contains(i * 2), "b gained value after a.or(b)");
+ }
+ assertEquals(a.cardinality(), 2000);
+ }
+
+ @Test
+ public void testOrCrossDirectionNoDeadlock() throws Exception {
+ // Concurrent A.or(B) and B.or(A) must not deadlock.
+ // Lock ordering by identityHashCode prevents the classic AB-BA
deadlock.
+ int pairs = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+ CountDownLatch latch = new CountDownLatch(pairs * 2);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ for (int i = 0; i < pairs; i++) {
+ final LongBitmap a = LongBitmaps.create();
+ final LongBitmap b = LongBitmaps.create();
+ for (int j = 0; j < 100; j++) {
+ a.add(j);
+ b.add(j + 50); // overlap
+ }
+ executor.submit(() -> {
+ try {
+ a.or(b);
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ executor.submit(() -> {
+ try {
+ b.or(a);
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS),
+ "or() cross-direction should not deadlock");
+ executor.shutdown();
+ assertEquals(errors.get(), 0);
+ }
+
+ @Test
+ public void testDrainToActionDoesNotHoldWriteLock() {
+ // Verify that the action runs outside the lock: a slow action should
not block
+ // concurrent operations. The select+remove phase holds writeLock
briefly, but
+ // the action invocation happens after unlock.
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 1000; i++) {
+ bitmap.add(i);
+ }
+
+ // Slow action: sleep briefly per value. drainTo should not hold any
lock
+ // during action execution — concurrent operations should remain
responsive.
+ long start = System.nanoTime();
+ long drained = bitmap.drainTo(100, v -> {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ }
+ });
+ long elapsed = System.nanoTime() - start;
+ assertEquals(drained, 100);
+ // Total elapsed ~100ms (sleep); the lock was released before action
ran.
+ assertTrue(elapsed >= 100_000_000L); // sanity: drainTo ran the
action 100 times
+ }
+
+ @Test
+ public void testUint32BoundaryRange() {
+ // Verify range APIs handle the uint32 upper boundary correctly.
+ // add(MAX_UINT32, MAX_UINT32+1) should add exactly one value:
MAX_UINT32.
+ // Note: MutableRoaringBitmap.contains(long, long) has a known issue
at this
+ // boundary where it returns false even when the value is present, so
we only
+ // test contains(long) single-value form and cardinality.
+ LongBitmap bitmap = LongBitmaps.create();
+ bitmap.add(0xFFFFFFFFL, 0x100000000L);
+ assertEquals(bitmap.cardinality(), 1);
+ assertTrue(bitmap.contains(0xFFFFFFFFL));
+
+ bitmap.remove(0xFFFFFFFFL, 0x100000000L);
+ assertEquals(bitmap.cardinality(), 0);
+ assertFalse(bitmap.contains(0xFFFFFFFFL));
+ }
+
+ @Test
+ public void testRangeReachesMaxUint32WithoutClamp() {
+ // Validates that add/remove do not need Math.min(to, UINT32_SIZE):
+ // validateRange(to - 1) already guarantees to <= UINT32_SIZE, and
RoaringBitmap
+ // handles to == UINT32_SIZE correctly at the boundary. If this test
ever fails,
+ // the clamp must be restored.
+ LongBitmap bitmap = LongBitmaps.create();
+
+ // Multi-value range that ends exactly at UINT32_SIZE: [MAX-1, MAX+1)
= {MAX-1, MAX}
+ bitmap.add(0xFFFFFFFEL, 0x100000000L);
+ assertEquals(bitmap.cardinality(), 2);
+ assertTrue(bitmap.contains(0xFFFFFFFEL));
+ assertTrue(bitmap.contains(0xFFFFFFFFL));
+
+ // Same range on remove
+ bitmap.remove(0xFFFFFFFEL, 0x100000000L);
+ assertEquals(bitmap.cardinality(), 0);
+ assertFalse(bitmap.contains(0xFFFFFFFEL));
+ assertFalse(bitmap.contains(0xFFFFFFFFL));
+
+ // Crossing container boundary (65535/65536) with to on a power-of-2
boundary
+ bitmap.add(65530L, 65540L);
+ assertEquals(bitmap.cardinality(), 10);
+ bitmap.remove(65530L, 65540L);
+ assertEquals(bitmap.cardinality(), 0);
+
+ // Out-of-range `to` must throw — proving validateRange guards the
upper bound.
+ assertThrows(IllegalArgumentException.class,
+ () -> bitmap.add(0L, 0x100000001L)); // to = UINT32_SIZE + 1
+ }
+
+ @Test
+ public void testSerializedSizeIsUpperBoundForSerialize() {
+ // serializedSize() is an upper bound (no runOptimize). serialize()
runs
+ // runOptimize, which only shrinks. So estimated >= actual.
+ long[] seeds = {0, 100, 65535, 65536, 1L << 30, 0xFFFFFF00L};
+ for (int trial = 0; trial < 5; trial++) {
+ LongBitmap bitmap = LongBitmaps.create();
+ java.util.Random rng = new java.util.Random(trial);
+ int count = 1000 + rng.nextInt(5000);
+ for (int i = 0; i < count; i++) {
+ bitmap.add(seeds[rng.nextInt(seeds.length)] +
rng.nextInt(100));
+ }
+ long estimated = bitmap.serializedSize();
+ byte[] actual = bitmap.serialize();
+ assertTrue(estimated >= actual.length,
+ "trial " + trial + ": estimated " + estimated + " < actual
" + actual.length);
+ }
+ }
+
+ @Test
+ public void testDrainToIsAtomic() {
+ // drainTo must atomically select+remove values. Concurrent add/remove
should
+ // not cause newly added values to be lost between snapshot and
removal.
+ LongBitmap bitmap = LongBitmaps.create();
+ for (int i = 0; i < 100; i++) {
+ bitmap.add(i);
+ }
+
+ // Drain with a slow action, while concurrently adding/removing values
+ AtomicInteger errors = new AtomicInteger(0);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ CountDownLatch latch = new CountDownLatch(2);
+
+ // T1: drain slowly
+ executor.submit(() -> {
+ try {
+ bitmap.drainTo(50, v -> {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ }
+ });
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+
+ // T2: add/remove while T1 drains
+ executor.submit(() -> {
+ try {
+ Thread.sleep(10); // let T1 start draining
+ for (int i = 0; i < 50; i++) {
+ bitmap.remove(i);
+ bitmap.add(i);
+ }
+ } catch (Exception e) {
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+
+ try {
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+ assertEquals(errors.get(), 0);
+ // After drain(50) and concurrent add/remove, the bitmap should
contain
+ // the values that were re-added by T2, not lost due to racy
andNot.
+ long finalCount = bitmap.cardinality();
+ // We drained ~50, then T2 added back some of those. Final count
>= 50
+ // (the un-drained values) is the key invariant.
+ assertTrue(finalCount >= 50,
+ "finalCount=" + finalCount + " should be >= 50 (un-drained
values)");
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
