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/01 18:11:00 UTC
(pulsar) 10/38: [fix][ml] Propagate cursor ledger deletion failures (#26734)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.2-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6b50c8645e9cbb2512310a773104af30ed26c0f8
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Tue Sep 29 03:21:41 2026 +0800
[fix][ml] Propagate cursor ledger deletion failures (#26734)
Co-authored-by: zhanghao <zh...@100.me>
(cherry picked from commit 0e75593fd392791b772b467a6a2f7adc63fa4e77)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 5 +-
.../mledger/impl/ManagedLedgerFactoryTest.java | 100 +++++++++++++++++++++
2 files changed, 103 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
index a719d484beb2..f131f484f8eb 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
@@ -1124,10 +1124,11 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
future.completeExceptionally(e);
}
});
+ }).exceptionally(ex -> {
+ future.completeExceptionally(ex);
+ return null;
});
-
-
return future;
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
index 32865fd4e6e7..c6410e4a2be3 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
@@ -18,12 +18,26 @@
*/
package org.apache.bookkeeper.mledger.impl;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
+import java.util.Optional;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteLedgerCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
@@ -31,6 +45,7 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.awaitility.Awaitility;
import org.testng.Assert;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
@@ -80,6 +95,91 @@ public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
assertEquals(mri.to.entryId, 0);
}
+ @Test(timeOut = 30000)
+ public void testDeleteCursorLedgerMetadataFailure() throws Exception {
+ String name = "delete-cursor-failure";
+ long cursorLedgerId = prepareClosedLedgerWithCursor(name);
+ // The ZooKeeper ledger manager reports metadata deletion failures (such as connection loss) as ZKException.
+ injectCursorDeleteFailure(cursorLedgerId, BKException.Code.ZKException);
+
+ CompletableFuture<Void> deletion = deleteAsync(name);
+ assertThatThrownBy(() -> deletion.get(5, TimeUnit.SECONDS))
+ .hasRootCauseInstanceOf(BKException.ZKException.class);
+ verify(bkc).asyncDeleteLedger(eq(cursorLedgerId), any(), any());
+ verify(metadataStore, never()).delete(eq("/managed-ledgers/" + name + "/c1"), any());
+ assertThat(factory.getManagedLedgerInfo(name).cursors).containsKey("c1");
+ }
+
+ @DataProvider(name = "cursorDeleteResults")
+ public Object[][] cursorDeleteResults() {
+ return new Object[][] {
+ {BKException.Code.OK},
+ {BKException.Code.NoSuchLedgerExistsException},
+ {BKException.Code.NoSuchLedgerExistsOnMetadataServerException}
+ };
+ }
+
+ @Test(dataProvider = "cursorDeleteResults", timeOut = 30000)
+ public void testDeleteCursorLedgerSuccess(int result) throws Exception {
+ String name = "delete-cursor-success";
+ long cursorLedgerId = prepareClosedLedgerWithCursor(name);
+ if (result != BKException.Code.OK) {
+ bkc.deleteLedger(cursorLedgerId);
+ injectCursorDeleteFailure(cursorLedgerId, result);
+ }
+
+ deleteAsync(name).get(5, TimeUnit.SECONDS);
+ verify(bkc).asyncDeleteLedger(eq(cursorLedgerId), any(), any());
+ verify(metadataStore).delete(eq("/managed-ledgers/" + name + "/c1"), eq(Optional.empty()));
+ assertThat(metadataStore.exists("/managed-ledgers/" + name).get(5, TimeUnit.SECONDS)).isFalse();
+ assertThat(bkc.getLedgers()).doesNotContain(cursorLedgerId);
+ }
+
+ private long prepareClosedLedgerWithCursor(String name) throws Exception {
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ bkc = spy(bkc);
+ metadataStore = spy(metadataStore);
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setMaxUnackedRangesToPersistInMetadataStore(0)
+ .setThrottleMarkDelete(0);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(name, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c1");
+ ledger.addEntry(new byte[] {1});
+ Position position = ledger.addEntry(new byte[] {2});
+ cursor.delete(position);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> cursor.getStats().getPersistLedgerSucceed() > 0);
+ ledger.close();
+ assertThat(factory.ledgers).doesNotContainKey(name);
+ long cursorLedgerId = factory.getManagedLedgerInfo(name).cursors.get("c1").cursorsLedgerId;
+ assertThat(cursorLedgerId).isNotEqualTo(-1L);
+ return cursorLedgerId;
+ }
+
+ private void injectCursorDeleteFailure(long cursorLedgerId, int result) {
+ doAnswer(invocation -> {
+ DeleteCallback callback = invocation.getArgument(1);
+ callback.deleteComplete(result, invocation.getArgument(2));
+ return null;
+ }).when(bkc).asyncDeleteLedger(eq(cursorLedgerId), any(), any());
+ }
+
+ private CompletableFuture<Void> deleteAsync(String name) {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ factory.asyncDelete(name, new DeleteLedgerCallback() {
+ @Override
+ public void deleteLedgerComplete(Object ctx) {
+ result.complete(null);
+ }
+
+ @Override
+ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ result.completeExceptionally(exception);
+ }
+ }, null);
+ return result;
+ }
+
/**
* see: https://github.com/apache/pulsar/pull/18688.
*/