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/21 19:45:40 UTC
(pulsar) 10/21: [fix][ml] Preserve individual batch ACKs when merging cumulative ACKs (#26998)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 75f479abb3d47c9b72802f07cb28500cb05fe215
Author: sinan liu <li...@gmail.com>
AuthorDate: Wed Oct 21 07:59:45 2026 +0800
[fix][ml] Preserve individual batch ACKs when merging cumulative ACKs (#26998)
(cherry picked from commit 7d8107eae889e0a8fe8ef69a1f9fd4fafa23ec22)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 27 ++--
.../impl/ManagedCursorBatchAckRecoveryTest.java | 136 +++++++++++++++++++++
2 files changed, 152 insertions(+), 11 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 39e9c26fac60..f3b1925d0236 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
@@ -2584,20 +2584,25 @@ public class ManagedCursorImpl implements ManagedCursor {
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;
+ // asyncDelete also updates these mutable bitmaps under the cursor write lock.
+ // Merge both ACK types without restoring individually acknowledged batch indexes.
+ lock.writeLock().lock();
+ try {
+ final var bitSet = batchDeletedIndexes.computeIfAbsent(position, __ -> givenBitSet);
+ bitSet.and(givenBitSet);
+ final Position newPosition;
+ if (bitSet.isEmpty()) {
+ batchDeletedIndexes.remove(position);
+ newPosition = PositionFactory.create(position.getLedgerId(), position.getEntryId());
} else {
- return v;
+ newPosition = ledger.getPreviousPosition(position);
}
- });
- final var newPosition = ledger.getPreviousPosition(position);
- batchDeletedIndexes.subMap(PositionFactory.EARLIEST, newPosition).clear();
- return newPosition;
+ batchDeletedIndexes.subMap(PositionFactory.EARLIEST, newPosition).clear();
+ return newPosition;
+ } finally {
+ lock.writeLock().unlock();
+ }
})
.orElse(position);
}
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 8a41c8ff8286..c2f7631ca886 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
@@ -20,11 +20,17 @@ package org.apache.bookkeeper.mledger.impl;
import static org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.defaultConfig;
import static org.assertj.core.api.Assertions.assertThat;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -54,6 +60,136 @@ public class ManagedCursorBatchAckRecoveryTest extends MockedBookKeeperTestCase
return new Object[][] {{false}, {true}};
}
+ @Test(dataProvider = "booleans")
+ public void testCumulativeAckPreservesIndividualBatchAcks(boolean metadataStore) throws Exception {
+ ManagedLedgerConfig config = defaultConfig()
+ .setMaxUnackedRangesToPersistInMetadataStore(metadataStore ? 20 : -1);
+ config.setDeletionAtBatchIndexLevelEnabled(true);
+ String name = "tenant/ns/persistent/mixed-batch-acks-" + metadataStore;
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ Position initial = ledger.addEntry(new byte[] {0});
+ Position batch = ledger.addEntry(new byte[] {1});
+ ledger.addEntry(new byte[] {2});
+ cursor.markDelete(initial);
+
+ // The individual ACK clears index 3; the cumulative ACK clears indexes 0 and 1.
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ batch.getLedgerId(), batch.getEntryId(), new long[] {7L}));
+ cursor.markDelete(AckSetStateUtil.createPositionWithAckSet(
+ batch.getLedgerId(), batch.getEntryId(), new long[] {12L}));
+ assertThat(cursor.getBatchPositionAckSet(batch)).containsExactly(4L);
+
+ // An older cumulative snapshot must not bring any acknowledged indexes back either.
+ cursor.markDelete(AckSetStateUtil.createPositionWithAckSet(
+ batch.getLedgerId(), batch.getEntryId(), new long[] {14L}));
+ assertThat(cursor.getBatchPositionAckSet(batch)).containsExactly(4L);
+ ledger.close();
+
+ @Cleanup
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("sub");
+ assertThat(recovered.getMarkDeletedPosition()).isEqualTo(initial);
+ assertThat(recovered.getBatchPositionAckSet(batch)).containsExactly(4L);
+ }
+
+ @Test(dataProvider = "booleans")
+ public void testCumulativeAckOverlapsIndividualBatchAck(boolean fullyAcknowledged) throws Exception {
+ ManagedLedgerConfig config = defaultConfig().setMaxUnackedRangesToPersistInMetadataStore(-1);
+ config.setDeletionAtBatchIndexLevelEnabled(true);
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "tenant/ns/persistent/concurrent-batch-acks-" + fullyAcknowledged, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ Position initial = ledger.addEntry(new byte[] {0});
+ Position batch = ledger.addEntry(new byte[] {1});
+ ledger.addEntry(new byte[] {2});
+ cursor.markDelete(initial);
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ batch.getLedgerId(), batch.getEntryId(), new long[] {14L}));
+
+ CountDownLatch individualMerged = new CountDownLatch(1);
+ CountDownLatch releaseIndividual = new CountDownLatch(1);
+ Position individualPosition = AckSetStateUtil.createPositionWithAckSet(
+ batch.getLedgerId(), batch.getEntryId(), new long[] {fullyAcknowledged ? 3L : 11L});
+ Iterable<Position> individualPositions = () -> new Iterator<>() {
+ private final Iterator<Position> delegate = List.of(individualPosition).iterator();
+
+ @Override
+ public boolean hasNext() {
+ boolean hasNext = delegate.hasNext();
+ if (!hasNext) {
+ // Pause after the real individual merge, while asyncDelete still holds the write lock.
+ individualMerged.countDown();
+ awaitLatch(releaseIndividual);
+ }
+ return hasNext;
+ }
+
+ @Override
+ public Position next() {
+ return delegate.next();
+ }
+ };
+ CountDownLatch cumulativeEntered = new CountDownLatch(1);
+ Position cumulativePosition = new AckSetPositionImpl(batch.getLedgerId(), batch.getEntryId(),
+ new long[] {12L}) {
+ @Override
+ public long[] getAckSet() {
+ cumulativeEntered.countDown();
+ return super.getAckSet();
+ }
+ };
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newSingleThreadExecutor(new DefaultThreadFactory("individual-batch-ack"));
+ Future<?> individual = executor.submit(() -> {
+ cursor.delete(individualPositions);
+ return null;
+ });
+ CompletableFuture<Void> cumulative = new CompletableFuture<>();
+ Thread cumulativeThread = new DefaultThreadFactory("cumulative-batch-ack").newThread(() -> {
+ try {
+ cursor.markDelete(cumulativePosition);
+ cumulative.complete(null);
+ } catch (Exception e) {
+ cumulative.completeExceptionally(e);
+ }
+ });
+ try {
+ assertThat(individualMerged.await(5, TimeUnit.SECONDS)).isTrue();
+ cumulativeThread.start();
+ assertThat(cumulativeEntered.await(5, TimeUnit.SECONDS)).isTrue();
+ // Both versions eventually wait for the cursor write lock. Before the fix, the cumulative
+ // path has already replaced the bitmap by this point, restoring individually ACKed bits.
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .until(() -> cumulativeThread.getState() == Thread.State.WAITING);
+ } finally {
+ releaseIndividual.countDown();
+ cumulativeThread.join(5000);
+ }
+ individual.get(5, TimeUnit.SECONDS);
+ cumulative.get(5, TimeUnit.SECONDS);
+
+ if (fullyAcknowledged) {
+ assertThat(cursor.getBatchPositionAckSet(batch)).isNull();
+ assertThat(cursor.isMessageDeleted(batch)).isTrue();
+ assertThat(cursor.getMarkDeletedPosition()).isEqualTo(batch);
+ } else {
+ assertThat(cursor.getBatchPositionAckSet(batch)).containsExactly(8L);
+ assertThat(cursor.getMarkDeletedPosition()).isEqualTo(initial);
+ }
+ }
+
+ private static void awaitLatch(CountDownLatch latch) {
+ try {
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
@Test
public void testFlushPersistsPartialBatchAckWithoutIndividualRanges() throws Exception {
ManagedLedgerConfig config = defaultConfig().setMaxUnackedRangesToPersistInMetadataStore(-1);