You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by da...@apache.org on 2025/05/23 01:38:51 UTC
(pulsar) branch master updated: [fix][ml] Fix ManagedCursorImpl.individualDeletedMessages concurrent issue (#24338)
This is an automated email from the ASF dual-hosted git repository.
daojun 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 376ae57e52a [fix][ml] Fix ManagedCursorImpl.individualDeletedMessages concurrent issue (#24338)
376ae57e52a is described below
commit 376ae57e52a3d9557ff26f5c184e0bb79b22fb37
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Fri May 23 09:38:45 2025 +0800
[fix][ml] Fix ManagedCursorImpl.individualDeletedMessages concurrent issue (#24338)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 6 ++++++
1 file changed, 6 insertions(+)
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 17988135c80..e73699564a2 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
@@ -642,6 +642,7 @@ public class ManagedCursorImpl implements ManagedCursor {
recoverIndividualDeletedMessages(positionInfo.getIndividualDeletedMessagesList());
} else if (positionInfo.getIndividualDeletedMessageRangesCount() > 0) {
List<LongListMap> rangeList = positionInfo.getIndividualDeletedMessageRangesList();
+ lock.writeLock().lock();
try {
Map<Long, long[]> rangeMap = rangeList.stream().collect(Collectors.toMap(LongListMap::getKey,
list -> list.getValuesList().stream().mapToLong(i -> i).toArray()));
@@ -664,6 +665,8 @@ public class ManagedCursorImpl implements ManagedCursor {
} catch (Exception e) {
log.warn("[{}]-{} Failed to recover individualDeletedMessages from serialized data", ledger.getName(),
name, e);
+ } finally {
+ lock.writeLock().unlock();
}
}
}
@@ -3259,10 +3262,13 @@ public class ManagedCursorImpl implements ManagedCursor {
* and deserialization error.
*/
if (getConfig().isUnackedRangesOpenCacheSetEnabled() && getConfig().isPersistIndividualAckAsLongArray()) {
+ lock.readLock().lock();
try {
internalRanges = individualDeletedMessages.toRanges(getConfig().getMaxUnackedRangesToPersist());
} catch (Exception e) {
log.warn("[{}]-{} Failed to serialize individualDeletedMessages", ledger.getName(), name, e);
+ } finally {
+ lock.readLock().unlock();
}
}
if (internalRanges != null && !internalRanges.isEmpty()) {