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/16 22:31:34 UTC
(pulsar) 11/24: [fix][ml] Wait for managed ledgers to close on factory shutdown, bounding each close (#26063)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2427500438bece99f2f409df6f7c9146abe1b527
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Oct 15 09:53:15 2026 -0700
[fix][ml] Wait for managed ledgers to close on factory shutdown, bounding each close (#26063)
(cherry picked from commit a1e83f5a70c316e7b3f82ff80208ebfbf83d70c7)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 133 ++++++++++++---------
.../mledger/impl/ManagedLedgerFactoryTest.java | 109 +++++++++++++++++
2 files changed, 184 insertions(+), 58 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 bef2fa078fea..da17023ea1b6 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
@@ -21,6 +21,7 @@ package org.apache.bookkeeper.mledger.impl;
import static org.apache.bookkeeper.mledger.ManagedLedgerException.getManagedLedgerException;
import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.AUTOMATIC_OFFLOAD_TRIGGER;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicates;
import com.google.common.collect.BoundType;
import com.google.common.collect.Maps;
@@ -29,6 +30,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.opentelemetry.api.OpenTelemetry;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
@@ -141,6 +143,9 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
//indicate whether shutdown() is called.
private volatile boolean closed;
+ // How long shutdown waits for each managed ledger to close
+ private long shutdownLedgerCloseTimeoutMillis =
+ TimeUnit.SECONDS.toMillis(ManagedLedgerImpl.AsyncOperationTimeoutSeconds);
/**
* Keep a flag to indicate whether we're currently connected to the metadata service.
@@ -591,68 +596,13 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
flushCursorsTask.cancel(true);
cacheEvictionExecutor.shutdownNow();
- List<String> ledgerNames = new ArrayList<>(this.ledgers.keySet());
- List<CompletableFuture<Void>> futures = new ArrayList<>(ledgerNames.size());
- int numLedgers = ledgerNames.size();
- log.info("Closing {} ledgers", numLedgers);
- for (String ledgerName : ledgerNames) {
- CompletableFuture<ManagedLedgerImpl> ledgerFuture = ledgers.remove(ledgerName);
- if (ledgerFuture == null) {
- continue;
- }
- CompletableFuture<Void> future = new CompletableFuture<>();
- futures.add(future);
- ledgerFuture.whenCompleteAsync((managedLedger, throwable) -> {
- if (throwable != null || managedLedger == null) {
- future.complete(null);
- return;
- }
- managedLedger.asyncClose(new AsyncCallbacks.CloseCallback() {
- @Override
- public void closeComplete(Object ctx) {
- future.complete(null);
- }
-
- @Override
- public void closeFailed(ManagedLedgerException exception, Object ctx) {
- log.warn("[{}] Got exception when closing managed ledger: {}", managedLedger.getName(),
- exception);
- future.complete(null);
- }
- }, null);
-
- }, scheduledExecutor.chooseThread());
- //close pendingInitializeManagedLedger directly to make sure all callbacks is called.
- PendingInitializeManagedLedger pendingLedger = pendingInitializeLedgers.get(ledgerName);
- if (pendingLedger != null && !ledgerFuture.isDone()) {
- ledgerFuture.completeExceptionally(new ManagedLedgerException.ManagedLedgerFactoryClosedException());
- }
- }
+ CompletableFuture<Void> closeLedgersFuture = closeLedgersOnShutdown();
CompletableFuture<BookKeeper> bookkeeperFuture = isBookkeeperManaged
? bookkeeperFactory.get()
: CompletableFuture.completedFuture(null);
return bookkeeperFuture
- .thenRun(() -> {
- log.info("Closing {} ledgers.", ledgers.size());
- //make sure all callbacks is called.
- ledgers.forEach(((ledgerName, ledgerFuture) -> {
- if (!ledgerFuture.isDone()) {
- ledgerFuture.completeExceptionally(
- new ManagedLedgerException.ManagedLedgerFactoryClosedException());
- } else {
- ManagedLedgerImpl managedLedger = ledgerFuture.getNow(null);
- if (managedLedger == null) {
- return;
- }
- try {
- managedLedger.close();
- } catch (Throwable throwable) {
- log.warn("[{}] Got exception when closing managed ledger: {}", managedLedger.getName(),
- throwable);
- }
- }
- }));
- }).whenCompleteAsync((__, ___) -> {
+ .thenCompose(__ -> closeLedgersFuture)
+ .whenCompleteAsync((__, ___) -> {
//wait for tasks in scheduledExecutor executed.
openTelemetryManagedCursorStats.close();
openTelemetryManagedLedgerStats.close();
@@ -669,6 +619,73 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
});
}
+ /**
+ * Closes all the managed ledgers of the factory, then the ones that opens racing with the shutdown added in the
+ * meantime, until none is left.
+ */
+ private CompletableFuture<Void> closeLedgersOnShutdown() {
+ List<String> ledgerNames = new ArrayList<>(ledgers.keySet());
+ if (ledgerNames.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ log.info("Closing {} ledgers", ledgerNames.size());
+ List<CompletableFuture<Void>> futures = new ArrayList<>(ledgerNames.size());
+ for (String ledgerName : ledgerNames) {
+ CompletableFuture<ManagedLedgerImpl> ledgerFuture = ledgers.remove(ledgerName);
+ if (ledgerFuture != null) {
+ futures.add(closeLedgerOnShutdown(ledgerName, ledgerFuture));
+ }
+ }
+ return Futures.waitForAll(futures).thenCompose(__ -> closeLedgersOnShutdown());
+ }
+
+ private CompletableFuture<Void> closeLedgerOnShutdown(String ledgerName,
+ CompletableFuture<ManagedLedgerImpl> ledgerFuture) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ ledgerFuture.whenCompleteAsync((managedLedger, throwable) -> {
+ if (throwable != null || managedLedger == null) {
+ future.complete(null);
+ return;
+ }
+ managedLedger.asyncClose(new AsyncCallbacks.CloseCallback() {
+ @Override
+ public void closeComplete(Object ctx) {
+ future.complete(null);
+ }
+
+ @Override
+ public void closeFailed(ManagedLedgerException exception, Object ctx) {
+ log.warn("[{}] Got exception when closing managed ledger: {}", managedLedger.getName(),
+ exception);
+ future.complete(null);
+ }
+ }, null);
+
+ }, scheduledExecutor.chooseThread());
+ //close pendingInitializeManagedLedger directly to make sure all callbacks is called.
+ PendingInitializeManagedLedger pendingLedger = pendingInitializeLedgers.get(ledgerName);
+ if (pendingLedger != null && !ledgerFuture.isDone()) {
+ ledgerFuture.completeExceptionally(new ManagedLedgerException.ManagedLedgerFactoryClosedException());
+ }
+ // Bound the wait, so that a ledger that never finishes opening or closing cannot hang the shutdown
+ return FutureUtil.addTimeoutHandling(future, Duration.ofMillis(shutdownLedgerCloseTimeoutMillis),
+ scheduledExecutor, () -> FutureUtil.createTimeoutException("Timed out closing managed ledger",
+ ManagedLedgerFactoryImpl.class, "closeLedgerOnShutdown"))
+ .exceptionally(ex -> {
+ log.warn("[{}] Timed out closing managed ledger on shutdown after {} ms", ledgerName,
+ shutdownLedgerCloseTimeoutMillis);
+ // Make sure that the callbacks of an open that is still in progress are called
+ ledgerFuture.completeExceptionally(
+ new ManagedLedgerException.ManagedLedgerFactoryClosedException());
+ return null;
+ });
+ }
+
+ @VisibleForTesting
+ void setShutdownLedgerCloseTimeoutMillis(long shutdownLedgerCloseTimeoutMillis) {
+ this.shutdownLedgerCloseTimeoutMillis = shutdownLedgerCloseTimeoutMillis;
+ }
+
@Override
public void shutdown() throws InterruptedException, ManagedLedgerException {
try {
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 c6410e4a2be3..4f8456e4d5c2 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
@@ -28,16 +28,24 @@ 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 static org.testng.Assert.assertTrue;
+import java.time.Duration;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteLedgerCallback;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenLedgerCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
@@ -214,6 +222,107 @@ public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
managedLedger2.close();
}
+ @Test(timeOut = 30000)
+ public void testShutdownWaitsForManagedLedgersToClose() throws Exception {
+ CompletableFuture<Void> closeRequested = new CompletableFuture<>();
+ CompletableFuture<Void> closeGate = new CompletableFuture<>();
+ replaceFactoryWithGatedClose("ml", closeRequested, closeGate);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("ml");
+
+ CompletableFuture<Void> shutdown = factory.shutdownAsync();
+ closeRequested.get(5, TimeUnit.SECONDS);
+ Awaitility.await().during(Duration.ofMillis(500)).atMost(Duration.ofSeconds(5))
+ .until(() -> !shutdown.isDone());
+
+ closeGate.complete(null);
+ shutdown.get(5, TimeUnit.SECONDS);
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.Closed);
+ }
+
+ @Test(timeOut = 30000)
+ public void testShutdownDoesNotWaitForeverForStuckManagedLedgerClose() throws Exception {
+ CompletableFuture<Void> closeRequested = new CompletableFuture<>();
+ replaceFactoryWithGatedClose("ml", closeRequested, new CompletableFuture<>());
+ factory.setShutdownLedgerCloseTimeoutMillis(100);
+ factory.open("ml");
+
+ // The close never completes, so the shutdown completes once it times out
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ assertTrue(closeRequested.isDone());
+ }
+
+ @Test(timeOut = 30000)
+ public void testShutdownFailsOpenThatNeverStartedInitializing() throws Exception {
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ // The BookKeeper client never becomes available, so the open never starts initializing the ledger
+ factory = new ManagedLedgerFactoryImpl(metadataStore, policyConfig -> new CompletableFuture<>(),
+ new ManagedLedgerFactoryConfig());
+ factory.setShutdownLedgerCloseTimeoutMillis(100);
+ CompletableFuture<ManagedLedger> open = new CompletableFuture<>();
+ factory.asyncOpen("ml", new OpenLedgerCallback() {
+ @Override
+ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
+ open.complete(ledger);
+ }
+
+ @Override
+ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ open.completeExceptionally(exception);
+ }
+ }, null);
+
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ assertThatThrownBy(() -> open.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerException.ManagedLedgerFactoryClosedException.class);
+ }
+
+ @Test(timeOut = 30000)
+ public void testShutdownClosesManagedLedgersAddedDuringShutdown() throws Exception {
+ CompletableFuture<Void> closeRequested = new CompletableFuture<>();
+ CompletableFuture<Void> closeGate = new CompletableFuture<>();
+ replaceFactoryWithGatedClose("ml", closeRequested, closeGate);
+ factory.open("ml");
+ ManagedLedgerImpl lateLedger = (ManagedLedgerImpl) factory.open("late-ml");
+ // Simulate an open that passed the closed check before the shutdown started, but added its ledger only
+ // after the shutdown had listed the ledgers to close
+ CompletableFuture<ManagedLedgerImpl> lateLedgerFuture = factory.ledgers.remove("late-ml");
+
+ CompletableFuture<Void> shutdown = factory.shutdownAsync();
+ closeRequested.get(5, TimeUnit.SECONDS);
+ factory.ledgers.put("late-ml", lateLedgerFuture);
+ closeGate.complete(null);
+
+ shutdown.get(5, TimeUnit.SECONDS);
+ assertEquals(lateLedger.getState(), ManagedLedgerImpl.State.Closed);
+ assertThat(factory.ledgers).isEmpty();
+ }
+
+ /**
+ * Replaces the factory with one whose managed ledger {@code gatedName} only closes once {@code closeGate}
+ * completes; {@code closeRequested} completes when its close is requested.
+ */
+ private void replaceFactoryWithGatedClose(String gatedName, CompletableFuture<Void> closeRequested,
+ CompletableFuture<Void> closeGate) throws Exception {
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc) {
+ @Override
+ protected ManagedLedgerImpl createManagedLedger(BookKeeper bk, MetaStore store, String name,
+ ManagedLedgerConfig config,
+ Supplier<CompletableFuture<Boolean>> mlOwnershipChecker) {
+ if (!name.equals(gatedName)) {
+ return super.createManagedLedger(bk, store, name, config, mlOwnershipChecker);
+ }
+ return new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker) {
+ @Override
+ public synchronized void asyncClose(CloseCallback callback, Object ctx) {
+ closeRequested.complete(null);
+ closeGate.thenRun(() -> super.asyncClose(callback, ctx));
+ }
+ };
+ }
+ };
+ }
+
private void waitManagedLedgerStateEquals(ManagedLedgerImpl managedLedger, ManagedLedgerImpl.State expectedStat){
Awaitility.await().untilAsserted(() ->
Assert.assertTrue(managedLedger.getState() == expectedStat));