diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java index a7442215264e4..01ed3f2e584e0 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java +++ b/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.LongArrayAckSets; 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 LongArrayAckSets.intersect(firstAckSet, secondAckSet); } public static boolean isAckSetEmpty(long[] ackSet) { - BitSetRecyclable bitSet = BitSetRecyclable.create().resetWords(ackSet); - boolean isEmpty = bitSet.isEmpty(); - bitSet.recycle(); - return isEmpty; + return LongArrayAckSets.cardinality(ackSet) == 0; } //This method is compare two position which position is bigger than another one. diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtilTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtilTest.java index d9c0c5a11eaee..4681e81240bda 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtilTest.java +++ b/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); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index fdad3200482d0..ad1efa69ba12e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/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; @@ -73,7 +72,7 @@ import org.apache.pulsar.common.stats.Rate; 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.common.util.LongArrayAckSets; import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; import org.apache.pulsar.transaction.common.exception.TransactionConflictException; @@ -389,7 +388,7 @@ public Future sendMessages(final List entries, long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); int remainingUnacked; if (ackSet != null) { - remainingUnacked = BitSet.valueOf(ackSet).cardinality(); + remainingUnacked = LongArrayAckSets.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 - LongArrayAckSets.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 = LongArrayAckSets.cardinality(cursorAckSet); + int currentCardinality = LongArrayAckSets.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 - LongArrayAckSets.cardinality(ackSets); } private void checkAckValidationError(CommandAck ack, Position position) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryBatchIndexesAcks.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryBatchIndexesAcks.java index 3126756b51469..b4061da09648b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryBatchIndexesAcks.java +++ b/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.LongArrayAckSets; @SuppressWarnings("unchecked") public class EntryBatchIndexesAcks { @@ -42,7 +42,7 @@ public int getTotalAckedIndexCount() { for (int i = 0; i < size; i++) { Pair pair = indexesAcks[i]; if (pair != null) { - count += pair.getLeft() - BitSet.valueOf(pair.getRight()).cardinality(); + count += pair.getLeft() - LongArrayAckSets.cardinality(pair.getRight()); } } return count; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java new file mode 100644 index 0000000000000..df7807e3946ee --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java @@ -0,0 +1,87 @@ +/* + * 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 sets held in the "long array" bit set representation — a long array + * containing a little-endian representation of all the bits in a bit set, as produced by + * {@link java.util.BitSet#toLongArray()} and accepted by {@link java.util.BitSet#valueOf(long[])} — without + * allocating a {@code BitSet} instance. + */ +public class LongArrayAckSets { + + private LongArrayAckSets() { + } + + /** + * Returns the number of bits set to {@code true} in the given ack set. + * + * @param ackSet a long array containing a little-endian representation of all the bits in a bit set + * @return the number of bits set to {@code true} + */ + public static int cardinality(long[] ackSet) { + int sum = 0; + for (long word : ackSet) { + sum += Long.bitCount(word); + } + return sum; + } + + /** + * Returns a new ack set whose words are the bitwise AND of the corresponding words of the two inputs. + * + *

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 ackSet1 a long array containing a little-endian representation of all the bits in a bit set + * @param ackSet2 a long array containing a little-endian representation of all the bits in a bit set + * @return a new array representing the intersection of the two ack sets, with trailing zero words trimmed + */ + public static long[] intersect(long[] ackSet1, long[] ackSet2) { + int len = Math.min(ackSet1.length, ackSet2.length); + while (len > 0 && (ackSet1[len - 1] & ackSet2[len - 1]) == 0) { + len--; + } + long[] result = new long[len]; + for (int i = 0; i < len; i++) { + result[i] = ackSet1[i] & ackSet2[i]; + } + return result; + } + + /** + * Returns the number of bits set to {@code true} after applying a logical AND to the given ack sets. + * + *

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 ackSet1 a long array containing a little-endian representation of all the bits in a bit set + * @param ackSet2 a long array containing a little-endian representation of all the bits in a bit set + * @return the number of bits set to {@code true} in {@code ackSet1 & ackSet2} + */ + public static int cardinalityOfIntersection(long[] ackSet1, long[] ackSet2) { + int sum = 0; + int len = Math.min(ackSet1.length, ackSet2.length); + for (int i = 0; i < len; i++) { + sum += Long.bitCount(ackSet1[i] & ackSet2[i]); + } + return sum; + } +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/BitSetRecyclable.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/BitSetRecyclable.java index b801d5f2b05a1..877b846da5965 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/BitSetRecyclable.java +++ b/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; } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java new file mode 100644 index 0000000000000..dbcc1eadc4b4a --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java @@ -0,0 +1,202 @@ +/* + * 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; + +import java.util.BitSet; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class LongArrayAckSetsTest { + + // ---- cardinality ---- + + @Test + public void testCardinalityEmptyArray() { + Assert.assertEquals(LongArrayAckSets.cardinality(new long[0]), 0); + } + + @Test + public void testCardinalityAllZero() { + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{0L, 0L, 0L}), 0); + } + + @Test + public void testCardinalityAllOnes() { + // -1L == all 64 bits set + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{-1L}), 64); + } + + @Test + public void testCardinalityMixed() { + // 0b1011 has 3 bits set; Long.MIN_VALUE (0x8000...0) has 1 bit set → total 4 + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{0b1011L, 0L, Long.MIN_VALUE}), 4); + } + + @Test + public void testCardinalityLongMaxValue() { + // Long.MAX_VALUE has 63 bits set + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{Long.MAX_VALUE}), 63); + } + + @Test + public void testCardinalityMultipleWords() { + // 0b1111 = 4 bits; 0b1010 = 2 bits → total 6 + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{0b1111L, 0b1010L}), 6); + } + + // ---- cardinalityOfIntersection ---- + + @Test + public void testAndCardinalityBothEmpty() { + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[0], new long[0]), 0); + } + + @Test + public void testAndCardinalityFirstEmpty() { + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[0], new long[]{Long.MAX_VALUE}), 0); + } + + @Test + public void testAndCardinalitySecondEmpty() { + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{Long.MAX_VALUE}, new long[0]), 0); + } + + @Test + public void testAndCardinalityNoOverlap() { + // 0b1010 & 0b0101 == 0 + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{0b1010L}, new long[]{0b0101L}), 0); + } + + @Test + public void testAndCardinalityFullOverlap() { + // -1L & -1L == -1L → 64 bits + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{-1L}, new long[]{-1L}), 64); + } + + @Test + public void testAndCardinalityPartialOverlap() { + // 0b1111 & 0b1010 == 0b1010 → 2 bits + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{0b1111L}, new long[]{0b1010L}), 2); + } + + @Test + public void testAndCardinalityFirstLonger() { + // Extra words in first array beyond second's length contribute 0 to AND result + // Long.MAX_VALUE & (nothing) == 0; 0b1010 & 0b1110 == 0b1010 → 2 bits + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( + new long[]{Long.MAX_VALUE, 0b1010L, Long.MAX_VALUE}, + new long[]{0L, 0b1110L}), 2); + } + + @Test + public void testAndCardinalitySecondLonger() { + // Extra words in second array beyond first's length contribute 0 + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( + new long[]{0b1110L}, + new long[]{0b1010L, Long.MAX_VALUE, Long.MAX_VALUE}), 2); + } + + @Test + public void testAndCardinalityMultipleWords() { + // word 0: 0b1111 & 0b1010 == 0b1010 → 2 bits + // word 1: -1L & 0b0011 == 0b0011 → 2 bits + // total: 4 + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( + new long[]{0b1111L, -1L}, + new long[]{0b1010L, 0b0011L}), 4); + } + + // ---- intersect ---- + + @Test + public void testIntersectBothEmpty() { + Assert.assertEquals(LongArrayAckSets.intersect(new long[0], new long[0]), new long[0]); + } + + @Test + public void testIntersectFirstEmpty() { + Assert.assertEquals(LongArrayAckSets.intersect(new long[0], new long[]{-1L}), new long[0]); + } + + @Test + public void testIntersectSecondEmpty() { + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{-1L}, new long[0]), new long[0]); + } + + @Test + public void testIntersectNoOverlap() { + // 0b1010 & 0b0101 == 0 → the all-zero word is trimmed, leaving an empty array + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[0]); + } + + @Test + public void testIntersectFullOverlap() { + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{-1L}, new long[]{-1L}), new long[]{-1L}); + } + + @Test + public void testIntersectPartialOverlap() { + // 0b1111 & 0b1010 == 0b1010 + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{0b1111L}, new long[]{0b1010L}), + new long[]{0b1010L}); + } + + @Test + public void testIntersectFirstLonger() { + // Result length = min(3, 2) = 2; extra word in first is dropped + Assert.assertEquals( + LongArrayAckSets.intersect(new long[]{0b1111L, 0b1010L, -1L}, new long[]{0b1100L, 0b1110L}), + new long[]{0b1100L, 0b1010L}); + } + + @Test + public void testIntersectSecondLonger() { + // Result length = min(2, 3) = 2; extra word in second is dropped + Assert.assertEquals( + LongArrayAckSets.intersect(new long[]{0b1111L, 0b1010L}, new long[]{0b1100L, 0b1110L, -1L}), + new long[]{0b1100L, 0b1010L}); + } + + @Test + public void testIntersectMultipleWords() { + Assert.assertEquals( + LongArrayAckSets.intersect(new long[]{0b1111L, -1L}, new long[]{0b1010L, 0b0011L}), + new long[]{0b1010L, 0b0011L}); + } + + @Test + public void testIntersectTrimsTrailingZeroWords() { + // High words AND to zero → result must be trimmed to the canonical form of BitSet#toLongArray + Assert.assertEquals( + LongArrayAckSets.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), + new long[]{-1L}); + // Same result as BitSet-based AND followed by toLongArray() + BitSet bitSet = BitSet.valueOf(new long[]{-1L, 0b01L}); + bitSet.and(BitSet.valueOf(new long[]{-1L, 0b10L})); + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), + bitSet.toLongArray()); + } + + @Test + public void testIntersectTrimsAllZeroResultToEmpty() { + Assert.assertEquals( + LongArrayAckSets.intersect(new long[]{0b1010L, 0b01L}, new long[]{0b0101L, 0b10L}), + new long[0]); + } +}