[improve][misc] Add LongArrayAckSets for allocation-free ack set operations

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java

@@ -21,6 +21,7 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.AckSetState;
 import org.apache.bookkeeper.mledger.impl.AckSetStateUtil;
+import org.apache.pulsar.common.util.AckSetUtil;
 import org.apache.pulsar.common.util.collections.BitSetRecyclable;
 
 public class PositionAckSetUtil {
@@ -54,20 +55,11 @@ public static void andAckSet(Position currentPosition, Position otherPosition) {
 
     //This method is do `and` operation for ack set
     public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) {
-        BitSetRecyclable thisAckSet = BitSetRecyclable.valueOf(firstAckSet);
-        BitSetRecyclable otherAckSet = BitSetRecyclable.valueOf(secondAckSet);
-        thisAckSet.and(otherAckSet);
-        long[] ackSet = thisAckSet.toLongArray();
-        thisAckSet.recycle();
-        otherAckSet.recycle();
-        return ackSet;
+        return AckSetUtil.intersect(firstAckSet, secondAckSet);
     }
 
     public static boolean isAckSetEmpty(long[] ackSet) {
-        BitSetRecyclable bitSet =  BitSetRecyclable.create().resetWords(ackSet);
-        boolean isEmpty = bitSet.isEmpty();
-        bitSet.recycle();
-        return isEmpty;
+        return AckSetUtil.cardinality(ackSet) == 0;
     }
 
     //This method is compare two position which position is bigger than another one.

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtilTest.java

@@ -49,6 +49,13 @@ public void isAckSetRepeatedTest() {
         assertFalse(isAckSetOverlap(thisBitSet.toLongArray(), otherBitSet.toLongArray()));
     }
 
+    @Test
+    public void andAckSetTrimsTrailingZeroWordsTest() {
+        // High words AND to zero → result must be trimmed like BitSet#toLongArray would
+        assertEquals(andAckSet(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), new long[]{-1L});
+        assertEquals(andAckSet(new long[]{0b01L}, new long[]{0b10L}), new long[0]);
+    }
+
     @Test
     public void compareToWithAckSetForCumulativeAckTest() {
         Position positionOne = PositionFactory.create(1, 1);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -32,7 +32,6 @@
 import it.unimi.dsi.fastutil.objects.ObjectIntPair;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.BitSet;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -71,9 +70,9 @@
 import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.common.stats.Rate;
+import org.apache.pulsar.common.util.AckSetUtil;
 import org.apache.pulsar.common.util.DateFormatter;
 import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.collections.BitSetRecyclable;
 import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
 import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
 
@@ -389,7 +388,7 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
                     long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i);
                     int remainingUnacked;
                     if (ackSet != null) {
-                        remainingUnacked = BitSet.valueOf(ackSet).cardinality();
+                        remainingUnacked = AckSetUtil.cardinality(ackSet);
                         unackedMessages -= (batchSize - remainingUnacked);
                     } else {
                         remainingUnacked = batchSize;
@@ -804,23 +803,15 @@ private long computeAckedCount(MessageIdData msgId, Position position, Consumer
         }
         long[] cursorAckSet = getCursorAckSet(position);
         if (cursorAckSet == null) {
-            return batchSize - BitSet.valueOf(ackSets).cardinality();
+            return batchSize - AckSetUtil.cardinality(ackSets);
         }
-        BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet);
-        int lastCardinality = cursorBitSet.cardinality();
-        BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(ackSets);
-        cursorBitSet.and(givenBitSet);
-        givenBitSet.recycle();
-        int currentCardinality = cursorBitSet.cardinality();
-        cursorBitSet.recycle();
+        int lastCardinality = AckSetUtil.cardinality(cursorAckSet);
+        int currentCardinality = AckSetUtil.cardinalityOfIntersection(cursorAckSet, ackSets);
         return lastCardinality - currentCardinality;
     }
 
     private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) {
-        BitSetRecyclable bitset = BitSetRecyclable.create().resetWords(ackSets);
-        long ackedCount = batchSize - bitset.cardinality();
-        bitset.recycle();
-        return ackedCount;
+        return batchSize - AckSetUtil.cardinality(ackSets);
     }
 
     private void checkAckValidationError(CommandAck ack, Position position) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryBatchIndexesAcks.java

@@ -20,8 +20,8 @@
 
 
 import io.netty.util.Recycler;
-import java.util.BitSet;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.util.AckSetUtil;
 
 @SuppressWarnings("unchecked")
 public class EntryBatchIndexesAcks {
@@ -42,7 +42,7 @@ public int getTotalAckedIndexCount() {
         for (int i = 0; i < size; i++) {
             Pair<Integer, long[]> pair = indexesAcks[i];
             if (pair != null) {
-                count += pair.getLeft() - BitSet.valueOf(pair.getRight()).cardinality();
+                count += pair.getLeft() - AckSetUtil.cardinality(pair.getRight());
             }
         }
         return count;

pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java

@@ -0,0 +1,81 @@
+/*
+ * 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;
+
+/**
+ * Utility methods for operating on ack-set word arrays without allocating a {@code BitSet} instance.
+ */
+public class AckSetUtil {
+
+    /**
+     * Returns the number of bits set to {@code true} in the given words.
+     *
+     * @param words a long array containing a little-endian representation of a sequence of bits
+     * @return the number of bits set to {@code true}
+     */
+    public static int cardinality(long[] words) {
+        int sum = 0;
+        for (long word : words) {
+            sum += Long.bitCount(word);
+        }
+        return sum;
+    }
+
+    /**
+     * Returns a new word array whose elements are the bitwise AND of the corresponding elements of the two inputs.
+     *
+     * <p>Extra words in the longer array are implicitly ANDed with zero and therefore contribute no set bits.
+     * Trailing all-zero words are trimmed from the result, so it is in the same canonical form produced by
+     * {@link java.util.BitSet#toLongArray()}.
+     *
+     * @param set1 a long array containing a little-endian representation of a sequence of bits
+     * @param set2 a long array containing a little-endian representation of a sequence of bits
+     * @return a new array representing the intersection of the two bit sets, with trailing zero words trimmed
+     */
+    public static long[] intersect(long[] set1, long[] set2) {
+        int len = Math.min(set1.length, set2.length);
+        while (len > 0 && (set1[len - 1] & set2[len - 1]) == 0) {
+            len--;
+        }
+        long[] result = new long[len];
+        for (int i = 0; i < len; i++) {
+            result[i] = set1[i] & set2[i];
+        }
+        return result;
+    }
+
+    /**
+     * Returns the number of bits set to {@code true} after applying a logical <b>AND</b> to the given words.
+     *
+     * <p>When the arrays differ in length, extra words in the longer array are treated as zero (i.e. the AND
+     * result for those positions is zero and contributes nothing to the cardinality).
+     *
+     * @param set1 a long array containing a little-endian representation of a sequence of bits
+     * @param set2 a long array containing a little-endian representation of a sequence of bits
+     * @return the number of bits set to {@code true} in {@code set1 & set2}
+     */
+    public static int cardinalityOfIntersection(long[] set1, long[] set2) {
+        int sum = 0;
+        int len = Math.min(set1.length, set2.length);
+        for (int i = 0; i < len; i++) {
+            sum += Long.bitCount(set1[i] & set2[i]);
+        }
+        return sum;
+    }
+}

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

@@ -872,8 +872,9 @@ public boolean intersects(BitSetRecyclable set) {
      */
     public int cardinality() {
         int sum = 0;
-        for (int i = 0; i < wordsInUse; i++)
+        for (int i = 0; i < wordsInUse; i++) {
             sum += Long.bitCount(words[i]);
+        }
         return sum;
     }