You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 22:31:37 UTC
(pulsar) 14/24: [fix][managed-ledger] Persist partial batch ACKs on flush (#26765)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3294aff538ea70d74541d04fcfd836d321a6d1cc
Author: Stochastic <13...@users.noreply.github.com>
AuthorDate: Fri Oct 16 02:36:53 2026 +0800
[fix][managed-ledger] Persist partial batch ACKs on flush (#26765)
(cherry picked from commit fa282309e22182fa54617fcd84de225dc52dec79)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 26 ++++++++--
.../impl/ManagedCursorBatchAckRecoveryTest.java | 55 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 3 deletions(-)
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 5ae744af54a5..6a7529c5d342 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
@@ -2581,6 +2581,7 @@ public class ManagedCursorImpl implements ManagedCursor {
lock.writeLock().lock();
boolean skipMarkDeleteBecauseAckedNothing = false;
+ boolean batchAckChanged = false;
try {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Deleting individual messages at {}. Current status: {} - md-position: {}",
@@ -2633,9 +2634,22 @@ public class ManagedCursorImpl implements ManagedCursor {
}
} else if (batchDeletedIndexes != null) {
final var givenBitSet = BitSet.valueOf(ackSet);
- final var bitSet = batchDeletedIndexes.computeIfAbsent(position, __ -> givenBitSet);
- if (givenBitSet != bitSet) {
- bitSet.and(givenBitSet);
+ // Check whether a bitmap for this batch already existed before computeIfAbsent,
+ // because a newly inserted bitmap must count as a change even though
+ // clone-and-AND-ing it with the same value yields no delta.
+ final boolean existedBefore = batchDeletedIndexes.containsKey(position);
+ // Use computeIfAbsent atomically — another thread may have inserted a bitmap
+ // between a separate get() and computeIfAbsent(), so merge based on whatever
+ // bitmap computeIfAbsent returns.
+ final var bitSet = batchDeletedIndexes.computeIfAbsent(position,
+ __ -> BitSet.valueOf(givenBitSet.toByteArray()));
+ final var previousBitSet = (BitSet) bitSet.clone();
+ bitSet.and(givenBitSet);
+ if (!existedBefore) {
+ // Newly inserted batch — state changed from "nothing" to a concrete bitmap.
+ batchAckChanged = true;
+ } else {
+ batchAckChanged |= !previousBitSet.equals(bitSet);
}
if (bitSet.isEmpty()) {
Position previousPosition = ledger.getPreviousPosition(position);
@@ -2697,6 +2711,12 @@ public class ManagedCursorImpl implements ManagedCursor {
} finally {
lock.writeLock().unlock();
if (skipMarkDeleteBecauseAckedNothing) {
+ if (batchAckChanged) {
+ // A partial batch ACK changes cursor state without creating an individual-delete range
+ // that would otherwise schedule a mark-delete write.
+ isDirty = true;
+ updateLastMarkDeleteEntryToLatest(markDeletePosition, null);
+ }
callback.deleteComplete(ctx);
}
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckRecoveryTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckRecoveryTest.java
index 6e4b952e9f7f..79bde5cdb6f8 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckRecoveryTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckRecoveryTest.java
@@ -19,6 +19,7 @@
package org.apache.bookkeeper.mledger.impl;
import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -31,8 +32,10 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.BatchedEntryDeletionIndexInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.metadata.api.Stat;
+import org.awaitility.Awaitility;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -50,6 +53,41 @@ public class ManagedCursorBatchAckRecoveryTest extends MockedBookKeeperTestCase
return new Object[][] {{false}, {true}};
}
+ @Test
+ public void testFlushPersistsPartialBatchAckWithoutIndividualRanges() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setMaxUnackedRangesToPersistInMetadataStore(-1);
+ config.setDeletionAtBatchIndexLevelEnabled(true);
+ config.setMaxBatchDeletedIndexToPersist(2);
+ String name = "tenant/ns/persistent/partial-batch-ack-flush";
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+
+ Position initialMarkDelete = ledger.addEntry(new byte[] {0});
+ Position partiallyAcked = ledger.addEntry(new byte[] {1});
+ ledger.addEntry(new byte[] {2});
+
+ cursor.delete(initialMarkDelete);
+ long persistedEntry = cursor.cursorLedger.getLastAddConfirmed();
+ assertThat(lastPersistedCursorPosition(cursor).getBatchedEntryDeletionIndexInfoCount()).isZero();
+
+ // The bitmap records the first message as acknowledged while leaving the second pending.
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ partiallyAcked.getLedgerId(), partiallyAcked.getEntryId(), new long[] {2L}));
+ assertThat(cursor.getIndividuallyDeletedMessagesSet().size()).isZero();
+
+ cursor.flush();
+
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .until(() -> cursor.cursorLedger.getLastAddConfirmed() > persistedEntry);
+ PositionInfo persisted = lastPersistedCursorPosition(cursor);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfoCount()).isEqualTo(1);
+ BatchedEntryDeletionIndexInfo ackInfo = persisted.getBatchedEntryDeletionIndexInfo(0);
+ assertThat(ackInfo.getPosition().getLedgerId()).isEqualTo(partiallyAcked.getLedgerId());
+ assertThat(ackInfo.getPosition().getEntryId()).isEqualTo(partiallyAcked.getEntryId());
+ assertThat(deleteSet(ackInfo)).containsExactly(2L);
+ }
+
@Test(dataProvider = "batchRecovery")
public void testPartialBatchAckPersistenceRetainsConfiguredCount(int batchCount, boolean metadataStore)
throws Exception {
@@ -200,4 +238,21 @@ public class ManagedCursorBatchAckRecoveryTest extends MockedBookKeeperTestCase
});
return result.get(5, TimeUnit.SECONDS);
}
+
+ private static PositionInfo lastPersistedCursorPosition(ManagedCursorImpl cursor) throws Exception {
+ CompletableFuture<PositionInfo> result = new CompletableFuture<>();
+ long entryId = cursor.cursorLedger.getLastAddConfirmed();
+ cursor.cursorLedger.asyncReadEntries(entryId, entryId, (rc, lh, entries, ctx) -> {
+ try {
+ if (rc != 0) {
+ throw new IllegalStateException("Failed to read cursor ledger entry: " + rc);
+ }
+ var entry = entries.nextElement();
+ result.complete(PositionInfo.parseFrom(entry.getEntry()));
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ }
+ }, null);
+ return result.get(5, TimeUnit.SECONDS);
+ }
}