You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2025/01/14 08:58:18 UTC
(pulsar) branch master updated: [fix][broker] Fix possible mark delete NPE when batch index ack is enabled (#23833)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new c92930f5441 [fix][broker] Fix possible mark delete NPE when batch index ack is enabled (#23833)
c92930f5441 is described below
commit c92930f544119d80e3451a5027694d56a2662ff2
Author: Yunze Xu <xy...@163.com>
AuthorDate: Tue Jan 14 16:58:11 2025 +0800
[fix][broker] Fix possible mark delete NPE when batch index ack is enabled (#23833)
---
.../bookkeeper/mledger/impl/AckSetState.java | 4 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 143 +++++++++------------
.../pendingack/PendingAckInMemoryDeleteTest.java | 7 +-
.../pendingack/impl/MLPendingAckStoreTest.java | 3 +-
4 files changed, 65 insertions(+), 92 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
index 363336e8311..6e9bf78b14c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/AckSetState.java
@@ -18,6 +18,8 @@
*/
package org.apache.bookkeeper.mledger.impl;
+import javax.annotation.Nullable;
+
/**
* Interface to manage the ackSet state attached to a position.
* Helpers in {@link AckSetStateUtil} to create positions with
@@ -28,7 +30,7 @@ public interface AckSetState {
* Get the ackSet bitset information encoded as a long array.
* @return the ackSet
*/
- long[] getAckSet();
+ @Nullable long[] getAckSet();
/**
* Set the ackSet bitset information as a long array.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 50f5f36b2d5..12cb22c161a 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -35,13 +35,14 @@ import com.google.protobuf.InvalidProtocolBufferException;
import java.time.Clock;
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -61,6 +62,8 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
+import javax.annotation.Nullable;
+import lombok.Getter;
import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
import org.apache.bookkeeper.client.AsyncCallback.OpenCallback;
import org.apache.bookkeeper.client.BKException;
@@ -105,7 +108,6 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
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.collections.LongPairRangeSet;
import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
import org.apache.pulsar.common.util.collections.LongPairRangeSet.RangeBoundConsumer;
@@ -200,7 +202,9 @@ public class ManagedCursorImpl implements ManagedCursor {
// Maintain the deletion status for batch messages
// (ledgerId, entryId) -> deletion indexes
- protected final ConcurrentSkipListMap<Position, BitSetRecyclable> batchDeletedIndexes;
+ @Getter
+ @VisibleForTesting
+ @Nullable protected final ConcurrentSkipListMap<Position, BitSet> batchDeletedIndexes;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private RateLimiter markDeleteLimiter;
@@ -709,6 +713,7 @@ public class ManagedCursorImpl implements ManagedCursor {
private void recoverBatchDeletedIndexes (
List<MLDataFormats.BatchedEntryDeletionIndexInfo> batchDeletedIndexInfoList) {
+ Objects.requireNonNull(batchDeletedIndexes);
lock.writeLock().lock();
try {
this.batchDeletedIndexes.clear();
@@ -720,8 +725,7 @@ public class ManagedCursorImpl implements ManagedCursor {
}
this.batchDeletedIndexes.put(
PositionFactory.create(batchDeletedIndexInfo.getPosition().getLedgerId(),
- batchDeletedIndexInfo.getPosition().getEntryId()),
- BitSetRecyclable.create().resetWords(array));
+ batchDeletedIndexInfo.getPosition().getEntryId()), BitSet.valueOf(array));
}
});
} finally {
@@ -1409,14 +1413,12 @@ public class ManagedCursorImpl implements ManagedCursor {
lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, isCompactionCursor()
? getProperties() : Collections.emptyMap(), null, null);
individualDeletedMessages.clear();
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- batchDeletedIndexes.values().forEach(BitSetRecyclable::recycle);
+ if (batchDeletedIndexes != null) {
batchDeletedIndexes.clear();
AckSetStateUtil.maybeGetAckSetState(newReadPosition).ifPresent(ackSetState -> {
long[] resetWords = ackSetState.getAckSet();
if (resetWords != null) {
- BitSetRecyclable ackSet = BitSetRecyclable.create().resetWords(resetWords);
- batchDeletedIndexes.put(newReadPosition, ackSet);
+ batchDeletedIndexes.put(newReadPosition, BitSet.valueOf(resetWords));
}
});
}
@@ -2045,47 +2047,7 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}] Mark delete cursor {} up to position: {}", ledger.getName(), name, position);
}
- Position newPosition = position;
-
- Optional<AckSetState> ackSetStateOptional = AckSetStateUtil.maybeGetAckSetState(newPosition);
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- if (ackSetStateOptional.isPresent()) {
- AtomicReference<BitSetRecyclable> bitSetRecyclable = new AtomicReference<>();
- BitSetRecyclable givenBitSet =
- BitSetRecyclable.create().resetWords(ackSetStateOptional.map(AckSetState::getAckSet).get());
- // In order to prevent the batch index recorded in batchDeletedIndexes from rolling back,
- // only update batchDeletedIndexes when the submitted batch index is greater
- // than the recorded index.
- batchDeletedIndexes.compute(newPosition,
- (k, v) -> {
- if (v == null) {
- return givenBitSet;
- }
- if (givenBitSet.nextSetBit(0) > v.nextSetBit(0)) {
- bitSetRecyclable.set(v);
- return givenBitSet;
- } else {
- bitSetRecyclable.set(givenBitSet);
- return v;
- }
- });
- if (bitSetRecyclable.get() != null) {
- bitSetRecyclable.get().recycle();
- }
- newPosition = ledger.getPreviousPosition(newPosition);
- }
- Map<Position, BitSetRecyclable> subMap = batchDeletedIndexes.subMap(PositionFactory.EARLIEST, newPosition);
- subMap.values().forEach(BitSetRecyclable::recycle);
- subMap.clear();
- } else {
- if (ackSetStateOptional.isPresent()) {
- AckSetState ackSetState = ackSetStateOptional.get();
- if (ackSetState.getAckSet() != null) {
- newPosition = ledger.getPreviousPosition(newPosition);
- }
- }
- }
-
+ Position newPosition = ackBatchPosition(position);
if (ledger.getLastConfirmedEntry().compareTo(newPosition) < 0) {
boolean shouldCursorMoveForward = false;
try {
@@ -2131,6 +2093,31 @@ public class ManagedCursorImpl implements ManagedCursor {
internalAsyncMarkDelete(newPosition, properties, callback, ctx);
}
+ private Position ackBatchPosition(Position position) {
+ return AckSetStateUtil.maybeGetAckSetState(position)
+ .map(AckSetState::getAckSet)
+ .map(ackSet -> {
+ if (batchDeletedIndexes == null) {
+ return ledger.getPreviousPosition(position);
+ }
+ // In order to prevent the batch index recorded in batchDeletedIndexes from rolling back,
+ // only update batchDeletedIndexes when the submitted batch index is greater
+ // than the recorded index.
+ final var givenBitSet = BitSet.valueOf(ackSet);
+ batchDeletedIndexes.compute(position, (k, v) -> {
+ if (v == null || givenBitSet.nextSetBit(0) > v.nextSetBit(0)) {
+ return givenBitSet;
+ } else {
+ return v;
+ }
+ });
+ final var newPosition = ledger.getPreviousPosition(position);
+ batchDeletedIndexes.subMap(PositionFactory.EARLIEST, newPosition).clear();
+ return newPosition;
+ })
+ .orElse(position);
+ }
+
protected void internalAsyncMarkDelete(final Position newPosition, Map<String, Long> properties,
final MarkDeleteCallback callback, final Object ctx) {
ledger.mbean.addMarkDeleteOp();
@@ -2236,12 +2223,10 @@ public class ManagedCursorImpl implements ManagedCursor {
try {
individualDeletedMessages.removeAtMost(mdEntry.newPosition.getLedgerId(),
mdEntry.newPosition.getEntryId());
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- Map<Position, BitSetRecyclable> subMap = batchDeletedIndexes.subMap(PositionFactory.EARLIEST,
+ if (batchDeletedIndexes != null) {
+ batchDeletedIndexes.subMap(PositionFactory.EARLIEST,
false, PositionFactory.create(mdEntry.newPosition.getLedgerId(),
- mdEntry.newPosition.getEntryId()), true);
- subMap.values().forEach(BitSetRecyclable::recycle);
- subMap.clear();
+ mdEntry.newPosition.getEntryId()), true).clear();
}
persistentMarkDeletePosition = mdEntry.newPosition;
} finally {
@@ -2376,11 +2361,8 @@ public class ManagedCursorImpl implements ManagedCursor {
}
if (internalIsMessageDeleted(position)) {
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
- if (bitSetRecyclable != null) {
- bitSetRecyclable.recycle();
- }
+ if (batchDeletedIndexes != null) {
+ batchDeletedIndexes.remove(position);
}
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Position was already deleted {}", ledger.getName(), name, position);
@@ -2389,11 +2371,8 @@ public class ManagedCursorImpl implements ManagedCursor {
}
long[] ackSet = AckSetStateUtil.getAckSetArrayOrNull(position);
if (ackSet == null) {
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
- if (bitSetRecyclable != null) {
- bitSetRecyclable.recycle();
- }
+ if (batchDeletedIndexes != null) {
+ batchDeletedIndexes.remove(position);
}
// Add a range (prev, pos] to the set. Adding the previous entry as an open limit to the range will
// make the RangeSet recognize the "continuity" between adjacent Positions.
@@ -2406,12 +2385,11 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}] [{}] Individually deleted messages: {}", ledger.getName(), name,
individualDeletedMessages);
}
- } else if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(ackSet);
- BitSetRecyclable bitSet = batchDeletedIndexes.computeIfAbsent(position, (v) -> givenBitSet);
+ } else if (batchDeletedIndexes != null) {
+ final var givenBitSet = BitSet.valueOf(ackSet);
+ final var bitSet = batchDeletedIndexes.computeIfAbsent(position, __ -> givenBitSet);
if (givenBitSet != bitSet) {
bitSet.and(givenBitSet);
- givenBitSet.recycle();
}
if (bitSet.isEmpty()) {
Position previousPosition = ledger.getPreviousPosition(position);
@@ -2419,10 +2397,7 @@ public class ManagedCursorImpl implements ManagedCursor {
previousPosition.getEntryId(),
position.getLedgerId(), position.getEntryId());
MSG_CONSUMED_COUNTER_UPDATER.incrementAndGet(this);
- BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
- if (bitSetRecyclable != null) {
- bitSetRecyclable.recycle();
- }
+ batchDeletedIndexes.remove(position);
}
}
}
@@ -3213,7 +3188,7 @@ public class ManagedCursorImpl implements ManagedCursor {
private List<MLDataFormats.BatchedEntryDeletionIndexInfo> buildBatchEntryDeletionIndexInfoList() {
lock.readLock().lock();
try {
- if (!getConfig().isDeletionAtBatchIndexLevelEnabled() || batchDeletedIndexes.isEmpty()) {
+ if (batchDeletedIndexes == null || batchDeletedIndexes.isEmpty()) {
return Collections.emptyList();
}
MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo
@@ -3221,9 +3196,9 @@ public class ManagedCursorImpl implements ManagedCursor {
MLDataFormats.BatchedEntryDeletionIndexInfo.Builder batchDeletedIndexInfoBuilder = MLDataFormats
.BatchedEntryDeletionIndexInfo.newBuilder();
List<MLDataFormats.BatchedEntryDeletionIndexInfo> result = new ArrayList<>();
- Iterator<Map.Entry<Position, BitSetRecyclable>> iterator = batchDeletedIndexes.entrySet().iterator();
+ final var iterator = batchDeletedIndexes.entrySet().iterator();
while (iterator.hasNext() && result.size() < getConfig().getMaxBatchDeletedIndexToPersist()) {
- Map.Entry<Position, BitSetRecyclable> entry = iterator.next();
+ final var entry = iterator.next();
nestedPositionBuilder.setLedgerId(entry.getKey().getLedgerId());
nestedPositionBuilder.setEntryId(entry.getKey().getEntryId());
batchDeletedIndexInfoBuilder.setPosition(nestedPositionBuilder.build());
@@ -3643,11 +3618,11 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public long[] getBatchPositionAckSet(Position position) {
if (batchDeletedIndexes != null) {
- BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.get(position);
- if (bitSetRecyclable == null) {
+ final var bitSet = batchDeletedIndexes.get(position);
+ if (bitSet == null) {
return null;
} else {
- return bitSetRecyclable.toLongArray();
+ return bitSet.toLongArray();
}
} else {
return null;
@@ -3750,8 +3725,8 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public long[] getDeletedBatchIndexesAsLongArray(Position position) {
- if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
- BitSetRecyclable bitSet = batchDeletedIndexes.get(position);
+ if (batchDeletedIndexes != null) {
+ final var bitSet = batchDeletedIndexes.get(position);
return bitSet == null ? null : bitSet.toLongArray();
} else {
return null;
@@ -3879,9 +3854,9 @@ public class ManagedCursorImpl implements ManagedCursor {
lock.readLock().unlock();
}
if (batchDeletedIndexes != null) {
- for (Map.Entry<Position, BitSetRecyclable> entry : this.batchDeletedIndexes.entrySet()) {
- BitSetRecyclable copiedBitSet = BitSetRecyclable.valueOf(entry.getValue());
- newNonDurableCursor.batchDeletedIndexes.put(entry.getKey(), copiedBitSet);
+ Objects.requireNonNull(newNonDurableCursor.batchDeletedIndexes);
+ for (final var entry : this.batchDeletedIndexes.entrySet()) {
+ newNonDurableCursor.batchDeletedIndexes.put(entry.getKey(), (BitSet) entry.getValue().clone());
}
}
return newNonDurableCursor;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
index 58cf59aa6b3..1b4fd451c10 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
@@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
@@ -45,7 +44,6 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.client.api.transaction.TxnID;
-import org.apache.pulsar.common.util.collections.BitSetRecyclable;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -223,10 +221,7 @@ public class PendingAckInMemoryDeleteTest extends TransactionTestBase {
(LinkedMap<TxnID, HashMap<Position, Position>>) field.get(pendingAckHandle);
assertTrue(individualAckOfTransaction.isEmpty());
managedCursor = (ManagedCursorImpl) testPersistentSubscription.getCursor();
- field = ManagedCursorImpl.class.getDeclaredField("batchDeletedIndexes");
- field.setAccessible(true);
- final ConcurrentSkipListMap<Position, BitSetRecyclable> batchDeletedIndexes =
- (ConcurrentSkipListMap<Position, BitSetRecyclable>) field.get(managedCursor);
+ final var batchDeletedIndexes = managedCursor.getBatchDeletedIndexes();
if (retryCnt == 0) {
//one message are not ack
Awaitility.await().until(() -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java
index 6dd3e6e7c78..6bbfb25ee2f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java
@@ -73,6 +73,7 @@ public class MLPendingAckStoreTest extends TransactionTestBase {
@BeforeClass
@Override
protected void setup() throws Exception {
+ conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
setUpBase(1, 1, NAMESPACE1 + "/test", 0);
}
@@ -304,4 +305,4 @@ public class MLPendingAckStoreTest extends TransactionTestBase {
}
return indexes;
}
-}
\ No newline at end of file
+}