[fix][misc] Remove RoaringBitmap dependency from pulsar-common (apache#23008)

(cherry picked from commit 32e29a3)
(cherry picked from commit 57a91b0)

Author: lhotari

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -379,8 +379,6 @@ The Apache Software License, Version 2.0
     - simpleclient_tracer_common-0.16.0.jar
     - simpleclient_tracer_otel-0.16.0.jar
     - simpleclient_tracer_otel_agent-0.16.0.jar
- * RoaringBitmap
-    - RoaringBitmap-1.2.0.jar
  * Log4J
     - log4j-api-2.18.0.jar
     - log4j-core-2.18.0.jar

managed-ledger/pom.xml

@@ -104,6 +104,10 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.roaringbitmap</groupId>
+      <artifactId>RoaringBitmap</artifactId>
+    </dependency>
     <dependency>
        <groupId>io.dropwizard.metrics</groupId>
        <artifactId>metrics-core</artifactId>

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/RangeSetWrapper.java

@@ -27,6 +27,7 @@
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet;
 import org.apache.pulsar.common.util.collections.OpenLongPairRangeSet;
+import org.roaringbitmap.RoaringBitSet;
 
 /**
  * Wraps other Range classes, and adds LRU, marking dirty data and other features on this basis.
@@ -55,7 +56,7 @@ public RangeSetWrapper(LongPairConsumer<T> rangeConverter,
         this.config = managedCursor.getManagedLedger().getConfig();
         this.rangeConverter = rangeConverter;
         this.rangeSet = config.isUnackedRangesOpenCacheSetEnabled()
-                ? new OpenLongPairRangeSet<>(rangeConverter)
+                ? new OpenLongPairRangeSet<>(rangeConverter, RoaringBitSet::new)
                 : new LongPairRangeSet.DefaultRangeSet<>(rangeConverter, rangeBoundConsumer);
         this.enableMultiEntry = config.isPersistentUnackedRangesWithMultipleEntriesEnabled();
     }

pulsar-common/pom.xml

@@ -243,11 +243,6 @@
       <artifactId>awaitility</artifactId>
       <scope>test</scope>
     </dependency>
-
-    <dependency>
-      <groupId>org.roaringbitmap</groupId>
-      <artifactId>RoaringBitmap</artifactId>
-    </dependency>
   </dependencies>
 
   <build>

pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/OpenLongPairRangeSet.java

@@ -28,9 +28,9 @@
 import java.util.NavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.apache.commons.lang.mutable.MutableInt;
-import org.roaringbitmap.RoaringBitSet;
 
 /**
  * A Concurrent set comprising zero or more ranges of type {@link LongPair}. This can be alternative of
@@ -48,6 +48,7 @@ public class OpenLongPairRangeSet<T extends Comparable<T>> implements LongPairRa
 
     protected final NavigableMap<Long, BitSet> rangeBitSetMap = new ConcurrentSkipListMap<>();
     private final LongPairConsumer<T> consumer;
+    private final Supplier<BitSet> bitSetSupplier;
 
     // caching place-holder for cpu-optimization to avoid calculating ranges again
     private volatile int cachedSize = 0;
@@ -56,7 +57,12 @@ public class OpenLongPairRangeSet<T extends Comparable<T>> implements LongPairRa
     private volatile boolean updatedAfterCachedForToString = true;
 
     public OpenLongPairRangeSet(LongPairConsumer<T> consumer) {
+        this(consumer, BitSet::new);
+    }
+
+    public OpenLongPairRangeSet(LongPairConsumer<T> consumer, Supplier<BitSet> bitSetSupplier) {
         this.consumer = consumer;
+        this.bitSetSupplier = bitSetSupplier;
     }
 
     /**
@@ -405,7 +411,7 @@ private int getSafeEntry(long value) {
     }
 
     private BitSet createNewBitSet() {
-        return new RoaringBitSet();
+        return bitSetSupplier.get();
     }
 
 }