From 5951c36b7e375cc04662e2d4e869c2411612877d Mon Sep 17 00:00:00 2001 From: Yike Xiao Date: Fri, 26 Jun 2026 00:29:28 +0800 Subject: [PATCH 1/4] [improve][common] Add cardinality helpers for bitset words Avoid creating BitSet or BitSetRecyclable instances when callers only need cardinality for long-array words. Add direct helpers for plain cardinality and AND cardinality, then use them in batch ack counting paths. Assisted-by: Codex --- .../mledger/util/PositionAckSetUtil.java | 5 +-- .../pulsar/broker/service/Consumer.java | 19 +++------- .../broker/service/EntryBatchIndexesAcks.java | 4 +- .../util/collections/BitSetRecyclable.java | 37 ++++++++++++++++++- .../BitSetRecyclableRecyclableTest.java | 16 ++++++++ 5 files changed, 60 insertions(+), 21 deletions(-) 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..4668673913d95 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 @@ -64,10 +64,7 @@ public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) { } public static boolean isAckSetEmpty(long[] ackSet) { - BitSetRecyclable bitSet = BitSetRecyclable.create().resetWords(ackSet); - boolean isEmpty = bitSet.isEmpty(); - bitSet.recycle(); - return isEmpty; + return BitSetRecyclable.cardinality(ackSet) == 0; } //This method is compare two position which position is bigger than another one. 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..cf9262e9d1761 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; @@ -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 = BitSetRecyclable.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 - BitSetRecyclable.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 = BitSetRecyclable.cardinality(cursorAckSet); + int currentCardinality = BitSetRecyclable.andCardinality(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 - BitSetRecyclable.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..2a546970e8f85 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.collections.BitSetRecyclable; @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() - BitSetRecyclable.cardinality(pair.getRight()); } } return count; 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..a10b5f1ea0422 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,43 @@ 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; + } + + /** + * 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 the number of bits set to {@code true} after applying a + * logical AND to the given words. + * + * @param firstWords the first long array containing a little-endian representation + * of a sequence of bits + * @param secondWords the second long array containing a little-endian representation + * of a sequence of bits + * @return the number of bits set to {@code true} in {@code firstWords & secondWords} + */ + public static int andCardinality(long[] firstWords, long[] secondWords) { + int sum = 0; + int wordsInUse = Math.min(firstWords.length, secondWords.length); + for (int i = 0; i < wordsInUse; i++) { + sum += Long.bitCount(firstWords[i] & secondWords[i]); + } return sum; } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java index 5bb615b19ecb7..2481f919ecf6a 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java @@ -65,4 +65,20 @@ public void testBitSetEmpty() { long[] array1 = bitSet.toLongArray(); Assert.assertTrue(BitSetRecyclable.create().resetWords(array1).isEmpty()); } + + @Test + public void testCardinalityOfWords() { + Assert.assertEquals(BitSetRecyclable.cardinality(new long[0]), 0); + Assert.assertEquals(BitSetRecyclable.cardinality(new long[] {0L, 0L}), 0); + Assert.assertEquals(BitSetRecyclable.cardinality(new long[] {0b1011L, 0L, Long.MIN_VALUE}), 4); + } + + @Test + public void testAndCardinalityOfWords() { + Assert.assertEquals(BitSetRecyclable.andCardinality(new long[0], new long[] {Long.MAX_VALUE}), 0); + Assert.assertEquals(BitSetRecyclable.andCardinality(new long[] {0b1111L}, new long[] {0b1010L}), 2); + Assert.assertEquals(BitSetRecyclable.andCardinality( + new long[] {Long.MAX_VALUE, 0b1010L, Long.MAX_VALUE}, + new long[] {0L, 0b1110L}), 2); + } } From 7ec3f9c7fc9ea58cfb6e6e1a906f72eafdb5d0de Mon Sep 17 00:00:00 2001 From: Yike Xiao Date: Wed, 1 Jul 2026 16:22:07 +0800 Subject: [PATCH 2/4] [improve][common] Introduce AckSetUtil for allocation-free ack-set word operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: avoid temporary BitSet/BitSetRecyclable allocations when callers only need cardinality or intersection on raw long-array words. Modifications: - Add AckSetUtil utility class in pulsar-common with: - cardinality(long[]) — popcount over a word array - intersect(long[], long[]) — element-wise AND, length = min of inputs - cardinalityOfIntersection(long[], long[]) — popcount of AND without allocating - Remove the two static helpers that were added to BitSetRecyclable in the previous commit and redirect all call sites (Consumer, EntryBatchIndexesAcks, PositionAckSetUtil) to AckSetUtil - Delegate PositionAckSetUtil.andAckSet(long[], long[]) to AckSetUtil.intersect - Rewrite PositionAckSetUtil.isAckSetOverlap with bitwise ops (~(a|b) != 0), eliminating the BitSetRecyclable allocations and flip passes - Add comprehensive unit tests in AckSetUtilTest Assisted-by: Claude Sonnet 4.6 --- .../mledger/util/PositionAckSetUtil.java | 30 ++- .../pulsar/broker/service/Consumer.java | 12 +- .../broker/service/EntryBatchIndexesAcks.java | 4 +- .../apache/pulsar/common/util/AckSetUtil.java | 80 ++++++++ .../util/collections/BitSetRecyclable.java | 34 ---- .../pulsar/common/util/AckSetUtilTest.java | 181 ++++++++++++++++++ .../BitSetRecyclableRecyclableTest.java | 16 -- 7 files changed, 281 insertions(+), 76 deletions(-) create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java 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 4668673913d95..a5494748caf41 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.AckSetUtil; import org.apache.pulsar.common.util.collections.BitSetRecyclable; public class PositionAckSetUtil { @@ -30,16 +31,15 @@ public static boolean isAckSetOverlap(long[] currentAckSet, long[] otherAckSet) if (currentAckSet == null || otherAckSet == null) { return false; } - - BitSetRecyclable currentBitSet = BitSetRecyclable.valueOf(currentAckSet); - BitSetRecyclable otherBitSet = BitSetRecyclable.valueOf(otherAckSet); - currentBitSet.flip(0, currentBitSet.size()); - otherBitSet.flip(0, otherBitSet.size()); - currentBitSet.and(otherBitSet); - boolean isAckSetRepeated = !currentBitSet.isEmpty(); - currentBitSet.recycle(); - otherBitSet.recycle(); - return isAckSetRepeated; + // A bit of 0 means "acked". Overlap exists when any position is acked in both sets, + // i.e. ~(a | b) != 0 for some word pair. + int len = Math.min(currentAckSet.length, otherAckSet.length); + for (int i = 0; i < len; i++) { + if (~(currentAckSet[i] | otherAckSet[i]) != 0) { + return true; + } + } + return false; } //This method is do `and` operation for position's ack set @@ -54,17 +54,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) { - return BitSetRecyclable.cardinality(ackSet) == 0; + return AckSetUtil.cardinality(ackSet) == 0; } //This method is compare two position which position is bigger than another one. 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 cf9262e9d1761..4a83f5a108546 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 @@ -70,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; @@ -388,7 +388,7 @@ public Future sendMessages(final List entries, long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); int remainingUnacked; if (ackSet != null) { - remainingUnacked = BitSetRecyclable.cardinality(ackSet); + remainingUnacked = AckSetUtil.cardinality(ackSet); unackedMessages -= (batchSize - remainingUnacked); } else { remainingUnacked = batchSize; @@ -803,15 +803,15 @@ private long computeAckedCount(MessageIdData msgId, Position position, Consumer } long[] cursorAckSet = getCursorAckSet(position); if (cursorAckSet == null) { - return batchSize - BitSetRecyclable.cardinality(ackSets); + return batchSize - AckSetUtil.cardinality(ackSets); } - int lastCardinality = BitSetRecyclable.cardinality(cursorAckSet); - int currentCardinality = BitSetRecyclable.andCardinality(cursorAckSet, ackSets); + int lastCardinality = AckSetUtil.cardinality(cursorAckSet); + int currentCardinality = AckSetUtil.cardinalityOfIntersection(cursorAckSet, ackSets); return lastCardinality - currentCardinality; } private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) { - return batchSize - BitSetRecyclable.cardinality(ackSets); + return batchSize - AckSetUtil.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 2a546970e8f85..494ce910eb894 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 @@ -21,7 +21,7 @@ import io.netty.util.Recycler; import org.apache.commons.lang3.tuple.Pair; -import org.apache.pulsar.common.util.collections.BitSetRecyclable; +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 pair = indexesAcks[i]; if (pair != null) { - count += pair.getLeft() - BitSetRecyclable.cardinality(pair.getRight()); + count += pair.getLeft() - AckSetUtil.cardinality(pair.getRight()); } } return count; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java new file mode 100644 index 0000000000000..e10af6d2332fe --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java @@ -0,0 +1,80 @@ +/* + * 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 lombok.experimental.UtilityClass; + +/** + * Utility methods for operating on ack-set word arrays without allocating a {@code BitSet} instance. + */ +@UtilityClass +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. + * + *

When the arrays differ in length, the result has {@code min(set1.length, set2.length)} words — extra words + * in the longer array are implicitly ANDed with zero and therefore contribute no set bits. + * + * @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 + */ + public static long[] intersect(long[] set1, long[] set2) { + int len = Math.min(set1.length, set2.length); + 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 AND to the given words. + * + *

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 firstWords & secondWords} + */ + 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; + } +} 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 a10b5f1ea0422..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 @@ -878,40 +878,6 @@ public int cardinality() { return sum; } - /** - * 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 the number of bits set to {@code true} after applying a - * logical AND to the given words. - * - * @param firstWords the first long array containing a little-endian representation - * of a sequence of bits - * @param secondWords the second long array containing a little-endian representation - * of a sequence of bits - * @return the number of bits set to {@code true} in {@code firstWords & secondWords} - */ - public static int andCardinality(long[] firstWords, long[] secondWords) { - int sum = 0; - int wordsInUse = Math.min(firstWords.length, secondWords.length); - for (int i = 0; i < wordsInUse; i++) { - sum += Long.bitCount(firstWords[i] & secondWords[i]); - } - return sum; - } - /** * Performs a logical AND of this target bit set with the * argument bit set. This bit set is modified so that each bit in it diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java new file mode 100644 index 0000000000000..eae3a7ab2a966 --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java @@ -0,0 +1,181 @@ +/* + * 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 org.testng.Assert; +import org.testng.annotations.Test; + +public class AckSetUtilTest { + + // ---- cardinality ---- + + @Test + public void testCardinalityEmptyArray() { + Assert.assertEquals(AckSetUtil.cardinality(new long[0]), 0); + } + + @Test + public void testCardinalityAllZero() { + Assert.assertEquals(AckSetUtil.cardinality(new long[]{0L, 0L, 0L}), 0); + } + + @Test + public void testCardinalityAllOnes() { + // -1L == all 64 bits set + Assert.assertEquals(AckSetUtil.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(AckSetUtil.cardinality(new long[]{0b1011L, 0L, Long.MIN_VALUE}), 4); + } + + @Test + public void testCardinalityLongMaxValue() { + // Long.MAX_VALUE has 63 bits set + Assert.assertEquals(AckSetUtil.cardinality(new long[]{Long.MAX_VALUE}), 63); + } + + @Test + public void testCardinalityMultipleWords() { + // 0b1111 = 4 bits; 0b1010 = 2 bits → total 6 + Assert.assertEquals(AckSetUtil.cardinality(new long[]{0b1111L, 0b1010L}), 6); + } + + // ---- cardinalityOfIntersection ---- + + @Test + public void testAndCardinalityBothEmpty() { + Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[0], new long[0]), 0); + } + + @Test + public void testAndCardinalityFirstEmpty() { + Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[0], new long[]{Long.MAX_VALUE}), 0); + } + + @Test + public void testAndCardinalitySecondEmpty() { + Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{Long.MAX_VALUE}, new long[0]), 0); + } + + @Test + public void testAndCardinalityNoOverlap() { + // 0b1010 & 0b0101 == 0 + Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{0b1010L}, new long[]{0b0101L}), 0); + } + + @Test + public void testAndCardinalityFullOverlap() { + // -1L & -1L == -1L → 64 bits + Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{-1L}, new long[]{-1L}), 64); + } + + @Test + public void testAndCardinalityPartialOverlap() { + // 0b1111 & 0b1010 == 0b1010 → 2 bits + Assert.assertEquals(AckSetUtil.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(AckSetUtil.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(AckSetUtil.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(AckSetUtil.cardinalityOfIntersection( + new long[]{0b1111L, -1L}, + new long[]{0b1010L, 0b0011L}), 4); + } + + // ---- intersect ---- + + @Test + public void testIntersectBothEmpty() { + Assert.assertEquals(AckSetUtil.intersect(new long[0], new long[0]), new long[0]); + } + + @Test + public void testIntersectFirstEmpty() { + Assert.assertEquals(AckSetUtil.intersect(new long[0], new long[]{-1L}), new long[0]); + } + + @Test + public void testIntersectSecondEmpty() { + Assert.assertEquals(AckSetUtil.intersect(new long[]{-1L}, new long[0]), new long[0]); + } + + @Test + public void testIntersectNoOverlap() { + // 0b1010 & 0b0101 == 0 + Assert.assertEquals(AckSetUtil.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[]{0L}); + } + + @Test + public void testIntersectFullOverlap() { + Assert.assertEquals(AckSetUtil.intersect(new long[]{-1L}, new long[]{-1L}), new long[]{-1L}); + } + + @Test + public void testIntersectPartialOverlap() { + // 0b1111 & 0b1010 == 0b1010 + Assert.assertEquals(AckSetUtil.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( + AckSetUtil.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( + AckSetUtil.intersect(new long[]{0b1111L, 0b1010L}, new long[]{0b1100L, 0b1110L, -1L}), + new long[]{0b1100L, 0b1010L}); + } + + @Test + public void testIntersectMultipleWords() { + Assert.assertEquals( + AckSetUtil.intersect(new long[]{0b1111L, -1L}, new long[]{0b1010L, 0b0011L}), + new long[]{0b1010L, 0b0011L}); + } +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java index 2481f919ecf6a..5bb615b19ecb7 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java @@ -65,20 +65,4 @@ public void testBitSetEmpty() { long[] array1 = bitSet.toLongArray(); Assert.assertTrue(BitSetRecyclable.create().resetWords(array1).isEmpty()); } - - @Test - public void testCardinalityOfWords() { - Assert.assertEquals(BitSetRecyclable.cardinality(new long[0]), 0); - Assert.assertEquals(BitSetRecyclable.cardinality(new long[] {0L, 0L}), 0); - Assert.assertEquals(BitSetRecyclable.cardinality(new long[] {0b1011L, 0L, Long.MIN_VALUE}), 4); - } - - @Test - public void testAndCardinalityOfWords() { - Assert.assertEquals(BitSetRecyclable.andCardinality(new long[0], new long[] {Long.MAX_VALUE}), 0); - Assert.assertEquals(BitSetRecyclable.andCardinality(new long[] {0b1111L}, new long[] {0b1010L}), 2); - Assert.assertEquals(BitSetRecyclable.andCardinality( - new long[] {Long.MAX_VALUE, 0b1010L, Long.MAX_VALUE}, - new long[] {0L, 0b1110L}), 2); - } } From 3667187ef84642dcebebc8e6651d32124164e97a Mon Sep 17 00:00:00 2001 From: Yike Xiao Date: Fri, 3 Jul 2026 18:51:32 +0800 Subject: [PATCH 3/4] Address review: trim trailing zero words in intersect, revert isAckSetOverlap - AckSetUtil.intersect now trims trailing all-zero words so the result stays in the canonical form produced by BitSet#toLongArray, matching the behavior of the previous andAckSet implementation - Revert the isAckSetOverlap rewrite entirely: the trailing-zero trimming semantics of the original implementation are questionable (without the batch size, trailing zero words are ambiguous), so leave the method untouched rather than re-encode that behavior here - Fix cardinalityOfIntersection javadoc param references - Add regression tests for intersect trailing-zero-word trimming Co-Authored-By: Claude Fable 5 --- .../mledger/util/PositionAckSetUtil.java | 19 +++++++------- .../mledger/util/PositionAckSetUtilTest.java | 7 ++++++ .../apache/pulsar/common/util/AckSetUtil.java | 17 +++++++------ .../pulsar/common/util/AckSetUtilTest.java | 25 +++++++++++++++++-- 4 files changed, 49 insertions(+), 19 deletions(-) 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 a5494748caf41..8b151b6988abf 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 @@ -31,15 +31,16 @@ public static boolean isAckSetOverlap(long[] currentAckSet, long[] otherAckSet) if (currentAckSet == null || otherAckSet == null) { return false; } - // A bit of 0 means "acked". Overlap exists when any position is acked in both sets, - // i.e. ~(a | b) != 0 for some word pair. - int len = Math.min(currentAckSet.length, otherAckSet.length); - for (int i = 0; i < len; i++) { - if (~(currentAckSet[i] | otherAckSet[i]) != 0) { - return true; - } - } - return false; + + BitSetRecyclable currentBitSet = BitSetRecyclable.valueOf(currentAckSet); + BitSetRecyclable otherBitSet = BitSetRecyclable.valueOf(otherAckSet); + currentBitSet.flip(0, currentBitSet.size()); + otherBitSet.flip(0, otherBitSet.size()); + currentBitSet.and(otherBitSet); + boolean isAckSetRepeated = !currentBitSet.isEmpty(); + currentBitSet.recycle(); + otherBitSet.recycle(); + return isAckSetRepeated; } //This method is do `and` operation for position's ack set 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-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java index e10af6d2332fe..30f1c1f46426e 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java @@ -18,12 +18,9 @@ */ package org.apache.pulsar.common.util; -import lombok.experimental.UtilityClass; - /** * Utility methods for operating on ack-set word arrays without allocating a {@code BitSet} instance. */ -@UtilityClass public class AckSetUtil { /** @@ -43,15 +40,19 @@ public static int cardinality(long[] words) { /** * Returns a new word array whose elements are the bitwise AND of the corresponding elements of the two inputs. * - *

When the arrays differ in length, the result has {@code min(set1.length, set2.length)} words — extra words - * in the longer array are implicitly ANDed with zero and therefore contribute no set bits. + *

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 + * @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]; @@ -65,9 +66,9 @@ public static long[] intersect(long[] set1, long[] set2) { *

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 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 firstWords & secondWords} + * @return the number of bits set to {@code true} in {@code set1 & set2} */ public static int cardinalityOfIntersection(long[] set1, long[] set2) { int sum = 0; diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java index eae3a7ab2a966..3f35434400df5 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.common.util; +import java.util.BitSet; import org.testng.Assert; import org.testng.annotations.Test; @@ -140,8 +141,8 @@ public void testIntersectSecondEmpty() { @Test public void testIntersectNoOverlap() { - // 0b1010 & 0b0101 == 0 - Assert.assertEquals(AckSetUtil.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[]{0L}); + // 0b1010 & 0b0101 == 0 → the all-zero word is trimmed, leaving an empty array + Assert.assertEquals(AckSetUtil.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[0]); } @Test @@ -178,4 +179,24 @@ public void testIntersectMultipleWords() { AckSetUtil.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( + AckSetUtil.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(AckSetUtil.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), + bitSet.toLongArray()); + } + + @Test + public void testIntersectTrimsAllZeroResultToEmpty() { + Assert.assertEquals( + AckSetUtil.intersect(new long[]{0b1010L, 0b01L}, new long[]{0b0101L, 0b10L}), + new long[0]); + } } From 4ad88cd387812dfa1fb2b8b15b410c6962ae27fb Mon Sep 17 00:00:00 2001 From: Yike Xiao Date: Mon, 6 Jul 2026 11:23:34 +0800 Subject: [PATCH 4/4] Address review: rename AckSetUtil to LongArrayAckSets - Rename the class and test per review: plural noun instead of Util suffix, named after the long[] word representation from BitSet#toLongArray; keep AckSet in the name since ack sets are the specific bit set usage this class serves - Generalize the javadoc around the long-array bit set representation with links to BitSet#toLongArray and BitSet#valueOf(long[]) - Rename parameters: words -> ackSet, set1/set2 -> ackSet1/ackSet2 - Add a private constructor so the class cannot be instantiated Co-Authored-By: Claude Fable 5 --- .../mledger/util/PositionAckSetUtil.java | 6 +- .../pulsar/broker/service/Consumer.java | 12 ++-- .../broker/service/EntryBatchIndexesAcks.java | 4 +- ...{AckSetUtil.java => LongArrayAckSets.java} | 48 +++++++++------- ...tilTest.java => LongArrayAckSetsTest.java} | 56 +++++++++---------- 5 files changed, 66 insertions(+), 60 deletions(-) rename pulsar-common/src/main/java/org/apache/pulsar/common/util/{AckSetUtil.java => LongArrayAckSets.java} (54%) rename pulsar-common/src/test/java/org/apache/pulsar/common/util/{AckSetUtilTest.java => LongArrayAckSetsTest.java} (63%) 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 8b151b6988abf..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,7 +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.LongArrayAckSets; import org.apache.pulsar.common.util.collections.BitSetRecyclable; public class PositionAckSetUtil { @@ -55,11 +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) { - return AckSetUtil.intersect(firstAckSet, secondAckSet); + return LongArrayAckSets.intersect(firstAckSet, secondAckSet); } public static boolean isAckSetEmpty(long[] ackSet) { - return AckSetUtil.cardinality(ackSet) == 0; + return LongArrayAckSets.cardinality(ackSet) == 0; } //This method is compare two position which position is bigger than another one. 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 4a83f5a108546..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 @@ -70,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.LongArrayAckSets; import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes; import org.apache.pulsar.transaction.common.exception.TransactionConflictException; @@ -388,7 +388,7 @@ public Future sendMessages(final List entries, long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); int remainingUnacked; if (ackSet != null) { - remainingUnacked = AckSetUtil.cardinality(ackSet); + remainingUnacked = LongArrayAckSets.cardinality(ackSet); unackedMessages -= (batchSize - remainingUnacked); } else { remainingUnacked = batchSize; @@ -803,15 +803,15 @@ private long computeAckedCount(MessageIdData msgId, Position position, Consumer } long[] cursorAckSet = getCursorAckSet(position); if (cursorAckSet == null) { - return batchSize - AckSetUtil.cardinality(ackSets); + return batchSize - LongArrayAckSets.cardinality(ackSets); } - int lastCardinality = AckSetUtil.cardinality(cursorAckSet); - int currentCardinality = AckSetUtil.cardinalityOfIntersection(cursorAckSet, ackSets); + int lastCardinality = LongArrayAckSets.cardinality(cursorAckSet); + int currentCardinality = LongArrayAckSets.cardinalityOfIntersection(cursorAckSet, ackSets); return lastCardinality - currentCardinality; } private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) { - return batchSize - AckSetUtil.cardinality(ackSets); + 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 494ce910eb894..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 @@ -21,7 +21,7 @@ import io.netty.util.Recycler; import org.apache.commons.lang3.tuple.Pair; -import org.apache.pulsar.common.util.AckSetUtil; +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() - AckSetUtil.cardinality(pair.getRight()); + count += pair.getLeft() - LongArrayAckSets.cardinality(pair.getRight()); } } return count; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java similarity index 54% rename from pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java rename to pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java index 30f1c1f46426e..df7807e3946ee 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/AckSetUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/LongArrayAckSets.java @@ -19,62 +19,68 @@ package org.apache.pulsar.common.util; /** - * Utility methods for operating on ack-set word arrays without allocating a {@code BitSet} instance. + * 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 AckSetUtil { +public class LongArrayAckSets { + + private LongArrayAckSets() { + } /** - * Returns the number of bits set to {@code true} in the given words. + * Returns the number of bits set to {@code true} in the given ack set. * - * @param words a long array containing a little-endian representation of a sequence of bits + * @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[] words) { + public static int cardinality(long[] ackSet) { int sum = 0; - for (long word : words) { + for (long word : ackSet) { 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. + * 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 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 + * @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[] set1, long[] set2) { - int len = Math.min(set1.length, set2.length); - while (len > 0 && (set1[len - 1] & set2[len - 1]) == 0) { + 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] = set1[i] & set2[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 words. + * 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 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} + * @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[] set1, long[] set2) { + public static int cardinalityOfIntersection(long[] ackSet1, long[] ackSet2) { int sum = 0; - int len = Math.min(set1.length, set2.length); + int len = Math.min(ackSet1.length, ackSet2.length); for (int i = 0; i < len; i++) { - sum += Long.bitCount(set1[i] & set2[i]); + sum += Long.bitCount(ackSet1[i] & ackSet2[i]); } return sum; } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java similarity index 63% rename from pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java rename to pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java index 3f35434400df5..dbcc1eadc4b4a 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/AckSetUtilTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/LongArrayAckSetsTest.java @@ -22,84 +22,84 @@ import org.testng.Assert; import org.testng.annotations.Test; -public class AckSetUtilTest { +public class LongArrayAckSetsTest { // ---- cardinality ---- @Test public void testCardinalityEmptyArray() { - Assert.assertEquals(AckSetUtil.cardinality(new long[0]), 0); + Assert.assertEquals(LongArrayAckSets.cardinality(new long[0]), 0); } @Test public void testCardinalityAllZero() { - Assert.assertEquals(AckSetUtil.cardinality(new long[]{0L, 0L, 0L}), 0); + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{0L, 0L, 0L}), 0); } @Test public void testCardinalityAllOnes() { // -1L == all 64 bits set - Assert.assertEquals(AckSetUtil.cardinality(new long[]{-1L}), 64); + 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(AckSetUtil.cardinality(new long[]{0b1011L, 0L, Long.MIN_VALUE}), 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(AckSetUtil.cardinality(new long[]{Long.MAX_VALUE}), 63); + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{Long.MAX_VALUE}), 63); } @Test public void testCardinalityMultipleWords() { // 0b1111 = 4 bits; 0b1010 = 2 bits → total 6 - Assert.assertEquals(AckSetUtil.cardinality(new long[]{0b1111L, 0b1010L}), 6); + Assert.assertEquals(LongArrayAckSets.cardinality(new long[]{0b1111L, 0b1010L}), 6); } // ---- cardinalityOfIntersection ---- @Test public void testAndCardinalityBothEmpty() { - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[0], new long[0]), 0); + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[0], new long[0]), 0); } @Test public void testAndCardinalityFirstEmpty() { - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[0], new long[]{Long.MAX_VALUE}), 0); + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[0], new long[]{Long.MAX_VALUE}), 0); } @Test public void testAndCardinalitySecondEmpty() { - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{Long.MAX_VALUE}, new long[0]), 0); + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{Long.MAX_VALUE}, new long[0]), 0); } @Test public void testAndCardinalityNoOverlap() { // 0b1010 & 0b0101 == 0 - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{0b1010L}, new long[]{0b0101L}), 0); + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{0b1010L}, new long[]{0b0101L}), 0); } @Test public void testAndCardinalityFullOverlap() { // -1L & -1L == -1L → 64 bits - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{-1L}, new long[]{-1L}), 64); + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection(new long[]{-1L}, new long[]{-1L}), 64); } @Test public void testAndCardinalityPartialOverlap() { // 0b1111 & 0b1010 == 0b1010 → 2 bits - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection(new long[]{0b1111L}, new long[]{0b1010L}), 2); + 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(AckSetUtil.cardinalityOfIntersection( + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( new long[]{Long.MAX_VALUE, 0b1010L, Long.MAX_VALUE}, new long[]{0L, 0b1110L}), 2); } @@ -107,7 +107,7 @@ public void testAndCardinalityFirstLonger() { @Test public void testAndCardinalitySecondLonger() { // Extra words in second array beyond first's length contribute 0 - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection( + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( new long[]{0b1110L}, new long[]{0b1010L, Long.MAX_VALUE, Long.MAX_VALUE}), 2); } @@ -117,7 +117,7 @@ public void testAndCardinalityMultipleWords() { // word 0: 0b1111 & 0b1010 == 0b1010 → 2 bits // word 1: -1L & 0b0011 == 0b0011 → 2 bits // total: 4 - Assert.assertEquals(AckSetUtil.cardinalityOfIntersection( + Assert.assertEquals(LongArrayAckSets.cardinalityOfIntersection( new long[]{0b1111L, -1L}, new long[]{0b1010L, 0b0011L}), 4); } @@ -126,34 +126,34 @@ public void testAndCardinalityMultipleWords() { @Test public void testIntersectBothEmpty() { - Assert.assertEquals(AckSetUtil.intersect(new long[0], new long[0]), new long[0]); + Assert.assertEquals(LongArrayAckSets.intersect(new long[0], new long[0]), new long[0]); } @Test public void testIntersectFirstEmpty() { - Assert.assertEquals(AckSetUtil.intersect(new long[0], new long[]{-1L}), new long[0]); + Assert.assertEquals(LongArrayAckSets.intersect(new long[0], new long[]{-1L}), new long[0]); } @Test public void testIntersectSecondEmpty() { - Assert.assertEquals(AckSetUtil.intersect(new long[]{-1L}, new long[0]), new long[0]); + 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(AckSetUtil.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[0]); + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{0b1010L}, new long[]{0b0101L}), new long[0]); } @Test public void testIntersectFullOverlap() { - Assert.assertEquals(AckSetUtil.intersect(new long[]{-1L}, new long[]{-1L}), new long[]{-1L}); + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{-1L}, new long[]{-1L}), new long[]{-1L}); } @Test public void testIntersectPartialOverlap() { // 0b1111 & 0b1010 == 0b1010 - Assert.assertEquals(AckSetUtil.intersect(new long[]{0b1111L}, new long[]{0b1010L}), + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{0b1111L}, new long[]{0b1010L}), new long[]{0b1010L}); } @@ -161,7 +161,7 @@ public void testIntersectPartialOverlap() { public void testIntersectFirstLonger() { // Result length = min(3, 2) = 2; extra word in first is dropped Assert.assertEquals( - AckSetUtil.intersect(new long[]{0b1111L, 0b1010L, -1L}, new long[]{0b1100L, 0b1110L}), + LongArrayAckSets.intersect(new long[]{0b1111L, 0b1010L, -1L}, new long[]{0b1100L, 0b1110L}), new long[]{0b1100L, 0b1010L}); } @@ -169,14 +169,14 @@ public void testIntersectFirstLonger() { public void testIntersectSecondLonger() { // Result length = min(2, 3) = 2; extra word in second is dropped Assert.assertEquals( - AckSetUtil.intersect(new long[]{0b1111L, 0b1010L}, new long[]{0b1100L, 0b1110L, -1L}), + LongArrayAckSets.intersect(new long[]{0b1111L, 0b1010L}, new long[]{0b1100L, 0b1110L, -1L}), new long[]{0b1100L, 0b1010L}); } @Test public void testIntersectMultipleWords() { Assert.assertEquals( - AckSetUtil.intersect(new long[]{0b1111L, -1L}, new long[]{0b1010L, 0b0011L}), + LongArrayAckSets.intersect(new long[]{0b1111L, -1L}, new long[]{0b1010L, 0b0011L}), new long[]{0b1010L, 0b0011L}); } @@ -184,19 +184,19 @@ public void testIntersectMultipleWords() { public void testIntersectTrimsTrailingZeroWords() { // High words AND to zero → result must be trimmed to the canonical form of BitSet#toLongArray Assert.assertEquals( - AckSetUtil.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), + 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(AckSetUtil.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), + Assert.assertEquals(LongArrayAckSets.intersect(new long[]{-1L, 0b01L}, new long[]{-1L, 0b10L}), bitSet.toLongArray()); } @Test public void testIntersectTrimsAllZeroResultToEmpty() { Assert.assertEquals( - AckSetUtil.intersect(new long[]{0b1010L, 0b01L}, new long[]{0b0101L, 0b10L}), + LongArrayAckSets.intersect(new long[]{0b1010L, 0b01L}, new long[]{0b0101L, 0b10L}), new long[0]); } }