You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/09 15:12:18 UTC
(pulsar) branch master updated: [fix][ml] Fix batch ACK index loss when recovering cursor from MetadataStore (#26474)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 5a47023fa3b2 [fix][ml] Fix batch ACK index loss when recovering cursor from MetadataStore (#26474)
5a47023fa3b2 is described below
commit 5a47023fa3b29a432649352c0d056a4d3f847203
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Wed Sep 9 23:12:11 2026 +0800
[fix][ml] Fix batch ACK index loss when recovering cursor from MetadataStore (#26474)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 5 +
.../impl/ManagedCursorBatchAckRecoveryTest.java | 203 +++++++++++++++++++++
...gedCursorBatchAckMetadataStoreRecoveryTest.java | 195 ++++++++++++++++++++
3 files changed, 403 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 05446b7ef1c3..9aee3b6491f3 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
@@ -582,6 +582,11 @@ public class ManagedCursorImpl implements ManagedCursor {
recoverIndividualDeletedMessages(info.getIndividualDeletedMessagesCount(),
info::getIndividualDeletedMessageAt);
}
+ if (getConfig().isDeletionAtBatchIndexLevelEnabled()
+ && info.getBatchedEntryDeletionIndexInfosCount() > 0) {
+ recoverBatchDeletedIndexes(info.getBatchedEntryDeletionIndexInfosCount(),
+ info::getBatchedEntryDeletionIndexInfoAt);
+ }
Map<String, Long> recoveredProperties = Collections.emptyMap();
if (info.getPropertiesCount() > 0) {
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
new file mode 100644
index 000000000000..210c6a02a7a8
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckRecoveryTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
+import org.apache.bookkeeper.mledger.proto.BatchedEntryDeletionIndexInfo;
+import org.apache.bookkeeper.mledger.proto.ManagedCursorInfo;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.apache.pulsar.metadata.api.Stat;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/** Verifies recovery of partial batch acknowledgments from cursor metadata. */
+@Test(timeOut = 30000)
+public class ManagedCursorBatchAckRecoveryTest extends MockedBookKeeperTestCase {
+
+ @DataProvider(name = "batchRecovery")
+ public Object[][] batchRecovery() {
+ return new Object[][] {{1, false}, {2, false}, {3, false}, {1, true}, {2, true}, {3, true}};
+ }
+
+ @DataProvider(name = "booleans")
+ public Object[][] booleans() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "batchRecovery")
+ public void testPartialBatchAckPersistenceRetainsConfiguredCount(int batchCount, boolean metadataStore)
+ throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig()
+ .setMaxUnackedRangesToPersistInMetadataStore(metadataStore ? 20 : -1);
+ config.setDeletionAtBatchIndexLevelEnabled(true);
+ config.setMaxBatchDeletedIndexToPersist(2);
+ String name = "tenant/ns/persistent/partial-batch-recovery-" + batchCount + "-" + metadataStore;
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ List<Position> positions = new ArrayList<>();
+ for (int i = 0; i <= batchCount; i++) {
+ positions.add(ledger.addEntry(new byte[] {(byte) i}));
+ }
+ long[][] ackSets = {{2L}, {Long.MIN_VALUE, 1L}, {5L, 0L, 3L}};
+ for (int i = 1; i < positions.size(); i++) {
+ Position position = positions.get(i);
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ position.getLedgerId(), position.getEntryId(), ackSets[i - 1]));
+ }
+
+ // Advance mark-delete, then close to persist and reload the actual durable state.
+ cursor.delete(positions.get(0));
+ ledger.close();
+ ManagedCursorInfo persisted = storedCursorInfo(ledger);
+ if (metadataStore) {
+ assertThat(persisted.getCursorsLedgerId()).isEqualTo(-1L);
+ int expectedPersistedCount = Math.min(batchCount, 2);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfosCount())
+ .as("the partial ACK records were actually written before recovery")
+ .isEqualTo(expectedPersistedCount);
+ for (int i = 0; i < expectedPersistedCount; i++) {
+ BatchedEntryDeletionIndexInfo indexInfo = persisted.getBatchedEntryDeletionIndexInfoAt(i);
+ assertThat(indexInfo.getPosition().getLedgerId()).isEqualTo(positions.get(i + 1).getLedgerId());
+ assertThat(indexInfo.getPosition().getEntryId()).isEqualTo(positions.get(i + 1).getEntryId());
+ assertThat(deleteSet(indexInfo)).containsExactly(ackSets[i]);
+ }
+ } else {
+ assertThat(persisted.getCursorsLedgerId()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Cleanup
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("sub");
+ assertThat(recovered).isNotSameAs(cursor);
+ for (int i = 1; i <= batchCount; i++) {
+ long[] recoveredAckSet = recovered.getBatchPositionAckSet(positions.get(i));
+ if (i <= 2) {
+ assertThat(recoveredAckSet).containsExactly(ackSets[i - 1]);
+ } else {
+ assertThat(recoveredAckSet).as("ACK state above the configured batch cap is truncated").isNull();
+ }
+ }
+ }
+
+ @Test
+ public void testMetadataStoreRecoveryPreservesAllCursorState() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig()
+ .setMaxUnackedRangesToPersistInMetadataStore(20);
+ config.setDeletionAtBatchIndexLevelEnabled(true);
+ String name = "tenant/ns/persistent/combined-metadata-store-recovery";
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ List<Position> positions = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ positions.add(ledger.addEntry(new byte[] {(byte) i}));
+ }
+
+ cursor.markDelete(positions.get(0), Map.of("mark-delete-property", 7L));
+ cursor.setCursorProperties(Map.of("cursor-property", "value")).get(5, TimeUnit.SECONDS);
+ cursor.delete(positions.get(2));
+ long[] ackSet = {Long.MIN_VALUE, 1L};
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ positions.get(4).getLedgerId(), positions.get(4).getEntryId(), ackSet));
+ ledger.close();
+
+ ManagedCursorInfo persisted = storedCursorInfo(ledger);
+ assertThat(persisted.getCursorsLedgerId()).isEqualTo(-1L);
+ assertThat(persisted.getIndividualDeletedMessagesCount()).isEqualTo(1);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfosCount()).isEqualTo(1);
+ assertThat(deleteSet(persisted.getBatchedEntryDeletionIndexInfoAt(0))).containsExactly(ackSet);
+
+ @Cleanup
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("sub");
+ assertThat(recovered).isNotSameAs(cursor);
+ assertThat(recovered.getMarkDeletedPosition()).isEqualTo(positions.get(0));
+ assertThat(recovered.getProperties()).containsExactlyEntriesOf(Map.of("mark-delete-property", 7L));
+ assertThat(recovered.getCursorProperties()).containsExactlyEntriesOf(Map.of("cursor-property", "value"));
+ assertThat(recovered.isMessageDeleted(positions.get(1))).isFalse();
+ assertThat(recovered.isMessageDeleted(positions.get(2))).isTrue();
+ assertThat(recovered.getBatchPositionAckSet(positions.get(4))).containsExactly(ackSet);
+ }
+
+ @Test(dataProvider = "booleans")
+ public void testMetadataStoreRecoveryHandlesMissingOrDisabledBatchIndexAck(boolean recoverBatchIndexAck)
+ throws Exception {
+ ManagedLedgerConfig writeConfig = new ManagedLedgerConfig()
+ .setMaxUnackedRangesToPersistInMetadataStore(20);
+ writeConfig.setDeletionAtBatchIndexLevelEnabled(!recoverBatchIndexAck);
+ String name = "tenant/ns/persistent/batch-metadata-compatibility-" + recoverBatchIndexAck;
+ @Cleanup
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, writeConfig);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ Position position = ledger.addEntry(new byte[] {1});
+ if (!recoverBatchIndexAck) {
+ cursor.delete(AckSetStateUtil.createPositionWithAckSet(
+ position.getLedgerId(), position.getEntryId(), new long[] {2L}));
+ }
+ ledger.close();
+
+ ManagedCursorInfo persisted = storedCursorInfo(ledger);
+ assertThat(persisted.getCursorsLedgerId()).isEqualTo(-1L);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfosCount())
+ .isEqualTo(recoverBatchIndexAck ? 0 : 1);
+
+ ManagedLedgerConfig recoveryConfig = new ManagedLedgerConfig()
+ .setMaxUnackedRangesToPersistInMetadataStore(20);
+ recoveryConfig.setDeletionAtBatchIndexLevelEnabled(recoverBatchIndexAck);
+ @Cleanup
+ ManagedLedgerImpl reopened = (ManagedLedgerImpl) factory.open(name, recoveryConfig);
+ ManagedCursorImpl recovered = (ManagedCursorImpl) reopened.openCursor("sub");
+ assertThat(recovered.getBatchPositionAckSet(position)).isNull();
+ }
+
+ private static long[] deleteSet(BatchedEntryDeletionIndexInfo indexInfo) {
+ long[] result = new long[indexInfo.getDeleteSetsCount()];
+ for (int i = 0; i < result.length; i++) {
+ result[i] = indexInfo.getDeleteSetAt(i);
+ }
+ return result;
+ }
+
+ private static ManagedCursorInfo storedCursorInfo(ManagedLedgerImpl ledger) throws Exception {
+ CompletableFuture<ManagedCursorInfo> result = new CompletableFuture<>();
+ ledger.getStore().asyncGetCursorInfo(ledger.getName(), "sub", new MetaStoreCallback<>() {
+ @Override
+ public void operationComplete(ManagedCursorInfo info, Stat stat) {
+ result.complete(info);
+ }
+
+ @Override
+ public void operationFailed(MetaStoreException exception) {
+ result.completeExceptionally(exception);
+ }
+ });
+ return result.get(5, TimeUnit.SECONDS);
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckMetadataStoreRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckMetadataStoreRecoveryTest.java
new file mode 100644
index 000000000000..cfb3ba32290d
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorBatchAckMetadataStoreRecoveryTest.java
@@ -0,0 +1,195 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
+import org.apache.bookkeeper.mledger.proto.ManagedCursorInfo;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.metadata.api.Stat;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class ManagedCursorBatchAckMetadataStoreRecoveryTest extends SharedPulsarBaseTest {
+
+ @Test(timeOut = 30000)
+ public void testSharedConsumerSkipsPartiallyAcknowledgedBatchAfterTopicUnload() throws Exception {
+ int batchSize = 4;
+ String topicName = newTopicName();
+ String subscriptionName = "shared-sub";
+ List<MessageId> sentMessageIds = new ArrayList<>(batchSize);
+
+ try (Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
+ .topic(topicName)
+ .enableBatching(true)
+ .batchingMaxMessages(batchSize)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .create();
+ Consumer<Integer> consumer = pulsarClient.newConsumer(Schema.INT32)
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .receiverQueueSize(batchSize)
+ .enableBatchIndexAcknowledgment(true)
+ .isAckReceiptEnabled(true)
+ .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS)
+ .subscribe()) {
+ List<CompletableFuture<MessageId>> sendFutures = new ArrayList<>(batchSize);
+ for (int i = 0; i < batchSize; i++) {
+ sendFutures.add(producer.sendAsync(i));
+ }
+ producer.flush();
+ for (CompletableFuture<MessageId> sendFuture : sendFutures) {
+ sentMessageIds.add(sendFuture.get(5, TimeUnit.SECONDS));
+ }
+ assertSingleBatch(sentMessageIds, batchSize);
+
+ List<Message<Integer>> messages = new ArrayList<>(batchSize);
+ try {
+ for (int i = 0; i < batchSize; i++) {
+ Message<Integer> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(message).as("message %s from the original batch", i).isNotNull();
+ assertThat(message.getValue()).isEqualTo(i);
+ messages.add(message);
+ }
+ consumer.acknowledge(messages.get(0));
+ consumer.acknowledge(messages.get(1));
+ } finally {
+ messages.forEach(Message::release);
+ }
+ }
+
+ PersistentTopic originalTopic = (PersistentTopic) getTopicReference(topicName).orElseThrow();
+ ManagedLedgerImpl originalLedger = (ManagedLedgerImpl) originalTopic.getManagedLedger();
+ ManagedCursorImpl originalCursor = cursor(originalTopic, subscriptionName);
+ MessageIdAdv batchMessageId = (MessageIdAdv) sentMessageIds.get(0);
+ Position batchPosition = PositionFactory.create(batchMessageId.getLedgerId(), batchMessageId.getEntryId());
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertThat(originalCursor.getBatchPositionAckSet(batchPosition)).isNotNull());
+ long[] ackSetBeforeUnload = originalCursor.getBatchPositionAckSet(batchPosition);
+
+ admin.topics().unload(topicName);
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .until(() -> getTopicReference(topicName).isEmpty());
+
+ ManagedCursorInfo persisted = storedCursorInfo(originalLedger, subscriptionName);
+ assertThat(persisted.getCursorsLedgerId()).isEqualTo(-1L);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfosCount())
+ .as("the partial batch ACK was written to MetadataStore before recovery")
+ .isEqualTo(1);
+ assertThat(persisted.getBatchedEntryDeletionIndexInfoAt(0).getPosition().getLedgerId())
+ .isEqualTo(batchPosition.getLedgerId());
+ assertThat(persisted.getBatchedEntryDeletionIndexInfoAt(0).getPosition().getEntryId())
+ .isEqualTo(batchPosition.getEntryId());
+
+ try (PulsarClient newClient = newPulsarClient();
+ Consumer<Integer> recoveredConsumer = newClient.newConsumer(Schema.INT32)
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .receiverQueueSize(batchSize)
+ .enableBatchIndexAcknowledgment(true)
+ .isAckReceiptEnabled(true)
+ .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS)
+ .subscribe()) {
+ PersistentTopic recoveredTopic = (PersistentTopic) getTopicReference(topicName).orElseThrow();
+ ManagedCursorImpl recoveredCursor = cursor(recoveredTopic, subscriptionName);
+ assertThat(recoveredCursor).as("topic unload must rebuild the cursor").isNotSameAs(originalCursor);
+ assertThat(recoveredCursor.getBatchPositionAckSet(batchPosition)).containsExactly(ackSetBeforeUnload);
+
+ Set<Integer> expectedValues = new HashSet<>(Set.of(2, 3));
+ for (int i = 0; i < 2; i++) {
+ Message<Integer> message = recoveredConsumer.receive(5, TimeUnit.SECONDS);
+ assertThat(message).as("unacknowledged message %s", i).isNotNull();
+ try {
+ assertThat(expectedValues.remove(message.getValue()))
+ .as("only unacknowledged messages should be delivered")
+ .isTrue();
+ recoveredConsumer.acknowledge(message);
+ } finally {
+ message.release();
+ }
+ }
+ assertThat(expectedValues).isEmpty();
+
+ Message<Integer> unexpected = recoveredConsumer.receive(1, TimeUnit.SECONDS);
+ try {
+ assertThat(unexpected).as("acknowledged batch indexes must not be redelivered").isNull();
+ } finally {
+ if (unexpected != null) {
+ unexpected.release();
+ }
+ }
+ }
+ }
+
+ private static void assertSingleBatch(List<MessageId> messageIds, int batchSize) {
+ assertThat(messageIds).hasSize(batchSize);
+ MessageIdAdv first = (MessageIdAdv) messageIds.get(0);
+ for (int i = 0; i < messageIds.size(); i++) {
+ MessageIdAdv current = (MessageIdAdv) messageIds.get(i);
+ assertThat(current.getLedgerId()).isEqualTo(first.getLedgerId());
+ assertThat(current.getEntryId()).isEqualTo(first.getEntryId());
+ assertThat(current.getBatchIndex()).isEqualTo(i);
+ }
+ }
+
+ private static ManagedCursorImpl cursor(PersistentTopic topic, String subscriptionName) {
+ PersistentSubscription subscription = topic.getSubscription(subscriptionName);
+ return (ManagedCursorImpl) subscription.getCursor();
+ }
+
+ private static ManagedCursorInfo storedCursorInfo(ManagedLedgerImpl ledger, String subscriptionName)
+ throws Exception {
+ CompletableFuture<ManagedCursorInfo> result = new CompletableFuture<>();
+ ledger.getStore().asyncGetCursorInfo(ledger.getName(), subscriptionName, new MetaStoreCallback<>() {
+ @Override
+ public void operationComplete(ManagedCursorInfo info, Stat stat) {
+ result.complete(info);
+ }
+
+ @Override
+ public void operationFailed(MetaStoreException exception) {
+ result.completeExceptionally(exception);
+ }
+ });
+ return result.get(5, TimeUnit.SECONDS);
+ }
+}