[improve][broker] Use RoaringBitmap in tracking individual acks to re…

…duce memory usage (apache#23006) (cherry picked from commit ed39c4d) (cherry picked from commit 278d679)
datastax · Jul 15, 2024 · b488f1f · b488f1f
1 parent 1560edb
commit b488f1f
Show file tree

Hide file tree

Showing 7 changed files with 14 additions and 18 deletions.
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -498,8 +498,7 @@ The Apache Software License, Version 2.0
   * RxJava
     - io.reactivex.rxjava3-rxjava-3.0.1.jar
   * RoaringBitmap
-    - org.roaringbitmap-RoaringBitmap-0.9.44.jar
-    - org.roaringbitmap-shims-0.9.44.jar
+    - org.roaringbitmap-RoaringBitmap-1.2.0.jar
 
 BSD 3-clause "New" or "Revised" License
  * Google auth library

diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -379,6 +379,8 @@ 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

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/RangeSetWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/RangeSetWrapper.java
@@ -55,7 +55,7 @@ public RangeSetWrapper(LongPairConsumer<T> rangeConverter,
         this.config = managedCursor.getManagedLedger().getConfig();
         this.rangeConverter = rangeConverter;
         this.rangeSet = config.isUnackedRangesOpenCacheSetEnabled()
-                ? new OpenLongPairRangeSet<>(4096, rangeConverter)
+                ? new OpenLongPairRangeSet<>(rangeConverter)
                 : new LongPairRangeSet.DefaultRangeSet<>(rangeConverter, rangeBoundConsumer);
         this.enableMultiEntry = config.isPersistentUnackedRangesWithMultipleEntriesEnabled();
     }

diff --git a/pom.xml b/pom.xml
@@ -307,7 +307,7 @@ flexible messaging model and an intuitive client API.</description>
     <j2objc-annotations.version>1.3</j2objc-annotations.version>
     <lightproto-maven-plugin.version>0.4</lightproto-maven-plugin.version>
     <dependency-check-maven.version>10.0.2</dependency-check-maven.version>
-    <roaringbitmap.version>0.9.44</roaringbitmap.version>
+    <roaringbitmap.version>1.2.0</roaringbitmap.version>
     <extra-enforcer-rules.version>1.6.1</extra-enforcer-rules.version>
     <oshi.version>6.4.0</oshi.version>
   </properties>

diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
@@ -243,6 +243,11 @@
       <artifactId>awaitility</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.roaringbitmap</groupId>
+      <artifactId>RoaringBitmap</artifactId>
+    </dependency>
   </dependencies>
 
   <build>

diff --git a/...-common/src/main/java/org/apache/pulsar/common/util/collections/OpenLongPairRangeSet.java b/...-common/src/main/java/org/apache/pulsar/common/util/collections/OpenLongPairRangeSet.java
@@ -30,6 +30,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 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
@@ -46,8 +47,6 @@
 public class OpenLongPairRangeSet<T extends Comparable<T>> implements LongPairRangeSet<T> {
 
     protected final NavigableMap<Long, BitSet> rangeBitSetMap = new ConcurrentSkipListMap<>();
-    private boolean threadSafe = true;
-    private final int bitSetSize;
     private final LongPairConsumer<T> consumer;
 
     // caching place-holder for cpu-optimization to avoid calculating ranges again
@@ -57,16 +56,6 @@ public class OpenLongPairRangeSet<T extends Comparable<T>> implements LongPairRa
     private volatile boolean updatedAfterCachedForToString = true;
 
     public OpenLongPairRangeSet(LongPairConsumer<T> consumer) {
-        this(1024, true, consumer);
-    }
-
-    public OpenLongPairRangeSet(int size, LongPairConsumer<T> consumer) {
-        this(size, true, consumer);
-    }
-
-    public OpenLongPairRangeSet(int size, boolean threadSafe, LongPairConsumer<T> consumer) {
-        this.threadSafe = threadSafe;
-        this.bitSetSize = size;
         this.consumer = consumer;
     }
 
@@ -416,7 +405,7 @@ private int getSafeEntry(long value) {
     }
 
     private BitSet createNewBitSet() {
-        return this.threadSafe ? new ConcurrentBitSet(bitSetSize) : new BitSet(bitSetSize);
+        return new RoaringBitSet();
     }
 
 }
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
@@ -477,7 +477,8 @@ The Apache Software License, Version 2.0
     - stream-2.9.5.jar
   * High Performance Primitive Collections for Java
     - hppc-0.9.1.jar
-
+  * RoaringBitmap
+    - RoaringBitmap-1.2.0.jar
 
 Protocol Buffers License
  * Protocol Buffers