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/17 20:20:05 UTC
(pulsar) 11/22: [fix][ml] Complete the original future after ledger deletion retries (#26742)
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 ea1f4b5a477bf64cc29cede1a2e1c05f7ed0086d
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Sat Oct 17 21:53:51 2026 +0800
[fix][ml] Complete the original future after ledger deletion retries (#26742)
Co-authored-by: zhanghao <zh...@100.me>
(cherry picked from commit 998b01b58fb60320cf85cb8379f3c3838121cd7e)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 12 +-
.../mledger/impl/ManagedLedgerDeleteRetryTest.java | 125 +++++++++++++++++++++
2 files changed, 134 insertions(+), 3 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 14fb0ec3a239..d5caaf4d81c2 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -55,6 +55,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -3754,7 +3755,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
- private CompletableFuture<Void> asyncDeleteLedger(long ledgerId, long retry) {
+ @VisibleForTesting
+ CompletableFuture<Void> asyncDeleteLedger(long ledgerId, long retry) {
CompletableFuture<Void> future = new CompletableFuture<>();
asyncDeleteLedgerWithRetry(future, ledgerId, retry);
return future;
@@ -3774,8 +3776,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
future.completeExceptionally(BKException.create(rc));
return;
}
- scheduledExecutor.schedule(() -> asyncDeleteLedger(ledgerId, retry - 1),
- DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ try {
+ scheduledExecutor.schedule(() -> asyncDeleteLedgerWithRetry(future, ledgerId, retry - 1),
+ DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC, TimeUnit.SECONDS);
+ } catch (RejectedExecutionException e) {
+ future.completeExceptionally(e);
+ }
} else {
log.debug().attr("ledgerId", ledgerId).log("Deleted ledger");
future.complete(null);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerDeleteRetryTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerDeleteRetryTest.java
new file mode 100644
index 000000000000..de19fe4da2d8
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerDeleteRetryTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.defaultConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.mockito.ArgumentMatchers;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ManagedLedgerDeleteRetryTest extends MockedBookKeeperTestCase {
+
+ @Test(timeOut = 15_000)
+ public void testOriginalFutureCompletesWhenRetrySchedulingIsRejected() throws Exception {
+ LedgerHandle handle = bkc.createLedger(BookKeeper.DigestType.CRC32C, new byte[0]);
+ long ledgerId = handle.getId();
+ handle.close();
+ OrderedScheduler retryScheduler = OrderedScheduler.newSchedulerBuilder()
+ .numThreads(1).name("delete-retry-rejected").build();
+ try {
+ ManagedLedgerImpl ledger = new ManagedLedgerImpl(factory, bkc, factory.getMetaStore(), defaultConfig(),
+ retryScheduler, "delete-retry-rejected");
+ retryScheduler.shutdownNow();
+ bkc.failNow(BKException.Code.WriteException);
+
+ CompletableFuture<Void> result = ledger.asyncDeleteLedger(ledgerId,
+ ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_RETRIES);
+
+ assertThatThrownBy(() -> result.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(RejectedExecutionException.class);
+ assertThat(bkc.getLedgers()).contains(ledgerId);
+ } finally {
+ retryScheduler.shutdownNow();
+ }
+ }
+
+ @DataProvider(name = "deleteFailures")
+ public Object[][] deleteFailures() {
+ return new Object[][] {{0, false}, {1, false}, {2, false}, {3, false}, {1, true}};
+ }
+
+ @Test(dataProvider = "deleteFailures", timeOut = 15_000)
+ public void testOriginalFutureCompletesAfterDeleteRetries(int failures, boolean deleteBeforeRetry)
+ throws Exception {
+ LedgerHandle handle = bkc.createLedger(BookKeeper.DigestType.CRC32C, new byte[0]);
+ long ledgerId = handle.getId();
+ handle.close();
+ OrderedScheduler retryScheduler = mock(OrderedScheduler.class);
+ AtomicInteger scheduledRetries = new AtomicInteger();
+ // Execute the real retry tasks without waiting for the production backoff. Support both overloads
+ // so the regression also exercises the original implementation, which schedules a Callable.
+ doAnswer(invocation -> {
+ scheduledRetries.incrementAndGet();
+ if (deleteBeforeRetry) {
+ // The first attempt failed, but the ledger is gone when the retry runs:
+ // PulsarMockBookKeeper will return NoSuchLedgerExistsException.
+ bkc.deleteLedger(ledgerId);
+ }
+ Runnable task = invocation.getArgument(0);
+ return executor.schedule(task, 0, TimeUnit.SECONDS);
+ }).when(retryScheduler).schedule(any(Runnable.class),
+ eq((long) ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC), eq(TimeUnit.SECONDS));
+ doAnswer(invocation -> {
+ scheduledRetries.incrementAndGet();
+ if (deleteBeforeRetry) {
+ bkc.deleteLedger(ledgerId);
+ }
+ Callable<?> task = invocation.getArgument(0);
+ return executor.schedule(task, 0, TimeUnit.SECONDS);
+ }).when(retryScheduler).schedule(ArgumentMatchers.<Callable<?>>any(),
+ eq((long) ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC), eq(TimeUnit.SECONDS));
+
+ ManagedLedgerImpl ledger = new ManagedLedgerImpl(factory, bkc, factory.getMetaStore(), defaultConfig(),
+ retryScheduler, "delete-retry");
+ for (int i = 0; i < failures; i++) {
+ bkc.failAfter(i, BKException.Code.WriteException);
+ }
+
+ CompletableFuture<Void> result = ledger.asyncDeleteLedger(ledgerId,
+ ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_RETRIES);
+ if (failures == ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_RETRIES) {
+ assertThatThrownBy(() -> result.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(BKException.BKWriteException.class);
+ assertThat(bkc.getLedgers()).contains(ledgerId);
+ } else {
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertThat(bkc.getLedgers()).doesNotContain(ledgerId));
+ result.get(5, TimeUnit.SECONDS);
+ }
+ assertThat(scheduledRetries.get()).isEqualTo(
+ Math.min(failures, ManagedLedgerImpl.DEFAULT_LEDGER_DELETE_RETRIES - 1));
+ }
+}