You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "zhanghaou (via GitHub)" <gi...@apache.org> on 2026/09/29 03:24:10 UTC
[PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
zhanghaou opened a new pull request, #26742:
URL: https://github.com/apache/pulsar/pull/26742
### Motivation
When a BookKeeper ledger deletion fails and is retried, `ManagedLedgerImpl.asyncDeleteLedgerWithRetry()` schedules another call to `asyncDeleteLedger()`, which creates a new `CompletableFuture`.
The retry result is therefore disconnected from the original future. Even if a subsequent attempt succeeds or exhausts all retries, the original future remains incomplete, preventing callers from observing the deletion outcome.
### Modifications
- Pass the original future to `asyncDeleteLedgerWithRetry()` when scheduling subsequent attempts, so success or final failure completes the same future.
- Make `asyncDeleteLedger(long, long)` package-private and annotate it with `@VisibleForTesting` to test the returned future without reflection.
- Add regression coverage for immediate success, success after one or two failures, and failure after exhausting all three attempts.
The retry limit and backoff interval remain unchanged.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
https://github.com/zhanghaou/pulsar/pull/10
This change added tests and can be verified as follows:
- Before the fix, the immediate-success case passed, while all three cases involving retries failed because the original future remained incomplete. The successful-retry cases also verified that the ledger had actually been deleted before waiting on the original future.
- After the fix, all four cases in `ManagedLedgerDeleteRetryTest` passed.
- All six existing tests in `OffloadLedgerDeleteTest` passed.
- `quickCheck` passed.
The regression tests inject failures through `PulsarMockBookKeeper` and execute the scheduled retry tasks without waiting for the production backoff.
```bash
./gradlew :managed-ledger:test \
--tests 'org.apache.bookkeeper.mledger.impl.ManagedLedgerDeleteRetryTest' \
--tests 'org.apache.bookkeeper.mledger.impl.OffloadLedgerDeleteTest' \
-PtestRetryCount=0 -PtestFailFast=false
./gradlew quickCheck
```
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26742:
URL: https://github.com/apache/pulsar/pull/26742#discussion_r4230951569
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3707,7 +3708,7 @@ private void asyncDeleteLedgerWithRetry(CompletableFuture<Void> future, long led
future.completeExceptionally(BKException.create(rc));
return;
}
- scheduledExecutor.schedule(() -> asyncDeleteLedger(ledgerId, retry - 1),
+ scheduledExecutor.schedule(() -> asyncDeleteLedgerWithRetry(future, ledgerId, retry - 1),
Review Comment:
The fix looks good. One edge case I noticed: `schedule()` can throw a `RejectedExecutionException` if the scheduler is shutting down. In that case, the original future would still be left incomplete.
Would it make sense to catch the rejection here and complete the future exceptionally? I don't think this needs to block the fix, but it would close another gap in the same retry path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26742:
URL: https://github.com/apache/pulsar/pull/26742#discussion_r4230957614
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerDeleteRetryTest.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.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 {
+
+ @DataProvider(name = "deleteFailures")
+ public Object[][] deleteFailures() {
+ return new Object[][] {{0}, {1}, {2}, {3}};
+ }
+
+ @Test(dataProvider = "deleteFailures", timeOut = 15_000)
+ public void testOriginalFutureCompletesAfterDeleteRetries(int failures) throws Exception {
+ 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();
+ 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();
+ 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");
+ LedgerHandle handle = bkc.createLedger(BookKeeper.DigestType.CRC32C, new byte[0]);
+ long ledgerId = handle.getId();
+ handle.close();
+ 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);
+ }
Review Comment:
The retry tests look good. Could we also cover a case where the first delete attempt fails and the retry returns `NoSuchLedgerExistsException`?
The ledger might already have been deleted even if the original request reported an error. It'd be useful to verify that the original future still completes successfully in that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
Posted by "zhanghaou (via GitHub)" <gi...@apache.org>.
zhanghaou commented on code in PR #26742:
URL: https://github.com/apache/pulsar/pull/26742#discussion_r4256598796
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -3707,7 +3708,7 @@ private void asyncDeleteLedgerWithRetry(CompletableFuture<Void> future, long led
future.completeExceptionally(BKException.create(rc));
return;
}
- scheduledExecutor.schedule(() -> asyncDeleteLedger(ledgerId, retry - 1),
+ scheduledExecutor.schedule(() -> asyncDeleteLedgerWithRetry(future, ledgerId, retry - 1),
Review Comment:
Thanks for pointing this out. This will make the code safer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
Posted by "zhanghaou (via GitHub)" <gi...@apache.org>.
zhanghaou commented on code in PR #26742:
URL: https://github.com/apache/pulsar/pull/26742#discussion_r4256602134
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerDeleteRetryTest.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.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 {
+
+ @DataProvider(name = "deleteFailures")
+ public Object[][] deleteFailures() {
+ return new Object[][] {{0}, {1}, {2}, {3}};
+ }
+
+ @Test(dataProvider = "deleteFailures", timeOut = 15_000)
+ public void testOriginalFutureCompletesAfterDeleteRetries(int failures) throws Exception {
+ 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();
+ 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();
+ 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");
+ LedgerHandle handle = bkc.createLedger(BookKeeper.DigestType.CRC32C, new byte[0]);
+ long ledgerId = handle.getId();
+ handle.close();
+ 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);
+ }
Review Comment:
Thanks, that makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Complete the original future after ledger deletion retries [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26742:
URL: https://github.com/apache/pulsar/pull/26742
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org