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/22 20:22:46 UTC
(pulsar) 04/04: [fix][ml] Fix races in managed ledger initialization timeout handling (#23470)
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 9fae4173978871edd029af409fb7b2a391afff05
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Oct 22 23:08:22 2026 +0300
[fix][ml] Fix races in managed ledger initialization timeout handling (#23470)
(cherry picked from commit 98743b158977b435253427372dcea46eeff070c6)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 120 +++++++++++-----
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 33 +++++
.../apache/bookkeeper/mledger/impl/MetaStore.java | 5 +-
.../bookkeeper/mledger/impl/MetaStoreImpl.java | 4 +-
.../mledger/impl/ShadowManagedLedgerImpl.java | 16 ++-
.../mledger/impl/ManagedLedgerFactoryTest.java | 160 +++++++++++++++++++++
.../bookkeeper/mledger/impl/MetaStoreImplTest.java | 17 +++
7 files changed, 315 insertions(+), 40 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 d351e51e4795..2995b0db619b 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
@@ -104,6 +104,7 @@ import org.apache.bookkeeper.mledger.util.Errors;
import org.apache.bookkeeper.mledger.util.Futures;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
@@ -167,10 +168,12 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
private static class PendingInitializeManagedLedger {
private final ManagedLedgerImpl ledger;
+ private final CompletableFuture<ManagedLedgerImpl> future;
private final long createTimeMs;
- PendingInitializeManagedLedger(ManagedLedgerImpl ledger) {
+ PendingInitializeManagedLedger(ManagedLedgerImpl ledger, CompletableFuture<ManagedLedgerImpl> future) {
this.ledger = ledger;
+ this.future = future;
this.createTimeMs = System.currentTimeMillis();
}
@@ -490,29 +493,38 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
.log("Attempted to open ledger in unusable"
+ " state. Removing from the map to"
+ " recreate it");
- ledgers.remove(name, existingFuture);
+ removeLedgerFuture(name, existingFuture);
}
} catch (Exception e) {
- // Unable to get the future
+ // Unable to get the future. Remove it, so that a failed open is retried.
log.warn().attr("managedLedger", name).exception(e)
.log("Got exception while trying to"
+ " retrieve ledger");
+ removeLedgerFuture(name, existingFuture);
}
} else {
PendingInitializeManagedLedger pendingLedger = pendingInitializeLedgers.get(name);
if (null != pendingLedger) {
long pendingMs = System.currentTimeMillis() - pendingLedger.createTimeMs;
- if (pendingMs > TimeUnit.SECONDS.toMillis(config.getMetadataOperationsTimeoutSeconds())) {
+ // Removing the pending entry claims the initialization, so that it can't complete or be timed
+ // out concurrently. The pending entry could belong to a newer open than the future read above,
+ // if the future was already replaced.
+ if (pendingMs > TimeUnit.SECONDS.toMillis(config.getMetadataOperationsTimeoutSeconds())
+ && pendingLedger.future == existingFuture
+ && pendingInitializeLedgers.remove(name, pendingLedger)) {
log.warn().attr("managedLedger", name)
.attr("pendingMs", pendingMs)
.log("Managed ledger has been pending in"
+ " initialize state too long,"
+ " remove it from cache to retry");
- ledgers.remove(name, existingFuture);
- pendingInitializeLedgers.remove(name, pendingLedger);
+ pendingLedger.ledger.abandonInitialization();
+ removeLedgerFuture(name, existingFuture);
+ // Fail the opens waiting for the timed out initialization. The managed ledger instance is
+ // closed once its initialization completes or fails.
+ existingFuture.completeExceptionally(new ManagedLedgerException(
+ "Managed ledger initialization timed out after " + pendingMs + " ms"));
}
}
-
}
}
@@ -527,15 +539,30 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
.thenAccept(bk -> {
final ManagedLedgerImpl newledger =
createManagedLedger(bk, store, name, config, mlOwnershipChecker);
- PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger);
+ PendingInitializeManagedLedger pendingLedger =
+ new PendingInitializeManagedLedger(newledger, future);
pendingInitializeLedgers.put(name, pendingLedger);
newledger.initialize(new ManagedLedgerInitializeLedgerCallback() {
@Override
public void initializeComplete() {
+ // The future is already done also if the factory shutdown timed out waiting for
+ // an open whose initialization hadn't started yet, so it couldn't be claimed
+ if (!pendingInitializeLedgers.remove(name, pendingLedger) || future.isDone()) {
+ // The initialization timed out or the factory was shut down while it was in
+ // progress, and the future was failed and removed from the map. Close the
+ // instance so that it doesn't keep running next to the one of a retried open.
+ log.warn().attr("managedLedger", name)
+ .attr("cost", System.currentTimeMillis() - startTimeOfCreation)
+ .log("Managed ledger initialization completed after it was abandoned,"
+ + " closing it");
+ closePendingInitializeLedger(name, newledger);
+ future.completeExceptionally(new ManagedLedgerException(
+ "Managed ledger initialization was abandoned"));
+ return;
+ }
log.info().attr("managedLedger", name)
.attr("cost", System.currentTimeMillis() - startTimeOfCreation)
.log("Successfully initialize managed ledger");
- pendingInitializeLedgers.remove(name, pendingLedger);
// May need to update the cursor position and wait them finished
newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger().whenComplete((__, ex) -> {
// ignore ex since it is handled in maybeUpdateCursorBeforeTrimmingConsumedLedger
@@ -555,24 +582,14 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
.log("Failed to initialize managed ledger");
}
- // Clean the map if initialization fails
- ledgers.remove(name, future);
- entryCacheManager.removeEntryCache(name);
-
- if (pendingInitializeLedgers.remove(name, pendingLedger)) {
- pendingLedger.ledger.asyncClose(new CloseCallback() {
- @Override
- public void closeComplete(Object ctx) {
- // no-op
- }
-
- @Override
- public void closeFailed(ManagedLedgerException exception, Object ctx) {
- log.warn().attr("managedLedger", name).exception(exception)
- .log("Failed to close a pending initialization managed ledger");
- }
- }, null);
- }
+ // Clean the map if initialization fails. The future is no longer in the map if the
+ // initialization timed out, and then the entry cache could already belong to the
+ // managed ledger of a retried open.
+ removeLedgerFuture(name, future);
+ // Close the instance also when the initialization timed out or the factory was shut
+ // down, since the instance is no longer closed elsewhere
+ pendingInitializeLedgers.remove(name, pendingLedger);
+ closePendingInitializeLedger(name, newledger);
future.completeExceptionally(e);
}
@@ -695,9 +712,44 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
return;
}
// Remove the ledger from the internal factory cache.
- if (ledgers.remove(ledger.getName(), ledgerFuture)) {
- entryCacheManager.removeEntryCache(ledger.getName());
- }
+ removeLedgerFuture(ledger.getName(), ledgerFuture);
+ }
+
+ /**
+ * Removes the future of a managed ledger from the map, and its entry cache with it, if the future is still mapped.
+ *
+ * <p>The entry cache is keyed by the managed ledger name, so it's removed atomically with the future. Otherwise a
+ * concurrent open could create a new managed ledger instance between the two removals, and the new instance would
+ * either get the entry cache of the old instance or have its own entry cache removed.
+ *
+ * @return true if the future was removed
+ */
+ private boolean removeLedgerFuture(String name, CompletableFuture<ManagedLedgerImpl> future) {
+ MutableBoolean removed = new MutableBoolean();
+ ledgers.computeIfPresent(name, (__, currentFuture) -> {
+ if (currentFuture != future) {
+ return currentFuture;
+ }
+ entryCacheManager.removeEntryCache(name);
+ removed.setTrue();
+ return null;
+ });
+ return removed.booleanValue();
+ }
+
+ private void closePendingInitializeLedger(String name, ManagedLedgerImpl ledger) {
+ ledger.asyncClose(new CloseCallback() {
+ @Override
+ public void closeComplete(Object ctx) {
+ // no-op
+ }
+
+ @Override
+ public void closeFailed(ManagedLedgerException exception, Object ctx) {
+ log.warn().attr("managedLedger", name).exception(exception)
+ .log("Failed to close a pending initialization managed ledger");
+ }
+ }, null);
}
public CompletableFuture<Void> shutdownAsync() throws ManagedLedgerException {
@@ -779,9 +831,13 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
}, null);
}, scheduledExecutor.chooseThread());
- //close pendingInitializeManagedLedger directly to make sure all callbacks is called.
+ // Fail an open whose initialization is in progress directly to make sure all callbacks are called. Removing
+ // the pending entry claims the initialization, so that the managed ledger instance is closed once its
+ // initialization completes.
PendingInitializeManagedLedger pendingLedger = pendingInitializeLedgers.get(ledgerName);
- if (pendingLedger != null && !ledgerFuture.isDone()) {
+ if (pendingLedger != null && pendingLedger.future == ledgerFuture
+ && pendingInitializeLedgers.remove(ledgerName, pendingLedger)) {
+ pendingLedger.ledger.abandonInitialization();
ledgerFuture.completeExceptionally(new ManagedLedgerException.ManagedLedgerFactoryClosedException());
}
// Bound the wait, so that a ledger that never finishes opening or closing cannot hang the shutdown
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 6eb83090644a..643afc6a4ebd 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
@@ -341,6 +341,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
AtomicReferenceFieldUpdater.newUpdater(ManagedLedgerImpl.class, State.class, "state");
protected volatile State state = null;
private volatile boolean migrated = false;
+ // Set when the factory abandons the initialization, so that it stops before its next step
+ private volatile boolean initializationAbandoned = false;
@Getter
private final OrderedScheduler scheduledExecutor;
@@ -467,6 +469,28 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
+ /**
+ * Abandons the initialization of this instance, because it timed out or the factory is shut down. The
+ * initialization fails before its next step, instead of fencing the ledgers or updating the metadata of an
+ * instance that a retried open initializes.
+ */
+ void abandonInitialization() {
+ initializationAbandoned = true;
+ }
+
+ /**
+ * Fails the initialization if it has been abandoned.
+ *
+ * @return true if the initialization was abandoned and failed
+ */
+ protected boolean failInitializationIfAbandoned(ManagedLedgerInitializeLedgerCallback callback) {
+ if (!initializationAbandoned) {
+ return false;
+ }
+ callback.initializeFailed(new ManagedLedgerException("Managed ledger initialization was abandoned"));
+ return true;
+ }
+
synchronized void initialize(final ManagedLedgerInitializeLedgerCallback callback, final Object ctx) {
log.info().attr("lazyCursorRecovery", config.isLazyCursorRecovery())
.attr("triggerOffloadOnTopicLoad", config.isTriggerOffloadOnTopicLoad())
@@ -477,6 +501,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
new MetaStoreCallback<ManagedLedgerInfo>() {
@Override
public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
+ if (failInitializationIfAbandoned(callback)) {
+ return;
+ }
ledgersStat = stat;
if (mlInfo.hasTerminatedPosition()) {
state = State.Terminated;
@@ -632,6 +659,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
protected synchronized void initializeBookKeeper(final ManagedLedgerInitializeLedgerCallback callback) {
+ if (failInitializationIfAbandoned(callback)) {
+ return;
+ }
log.debug().attr("ledgers", ledgers).log("Initializing bookkeeper");
// Calculate total entries and size
@@ -729,6 +759,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
protected void initializeCursors(final ManagedLedgerInitializeLedgerCallback callback) {
+ if (failInitializationIfAbandoned(callback)) {
+ return;
+ }
log.debug("Initializing cursors");
store.getCursors(name, new MetaStoreCallback<List<String>>() {
@Override
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java
index 4a28d24c925f..22b8c8a4d58e 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java
@@ -83,10 +83,11 @@ public interface MetaStore {
void watchManagedLedgerInfo(String ledgerName, UpdateCallback<ManagedLedgerInfo> callback);
/**
- * Unwatch the metadata changes for ledgerName.
+ * Unwatch the metadata changes for ledgerName, if the callback is still the one watching them.
* @param ledgerName
+ * @param callback the callback that was passed to {@link #watchManagedLedgerInfo(String, UpdateCallback)}
*/
- void unwatchManagedLedgerInfo(String ledgerName);
+ void unwatchManagedLedgerInfo(String ledgerName, UpdateCallback<ManagedLedgerInfo> callback);
/**
*
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
index 44e47323f909..5961e49ad581 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
@@ -357,8 +357,8 @@ public class MetaStoreImpl implements MetaStore, Consumer<Notification> {
}
@Override
- public void unwatchManagedLedgerInfo(String ledgerName) {
- managedLedgerInfoUpdateCallbackMap.remove(PREFIX + ledgerName);
+ public void unwatchManagedLedgerInfo(String ledgerName, UpdateCallback<ManagedLedgerInfo> callback) {
+ managedLedgerInfoUpdateCallbackMap.remove(PREFIX + ledgerName, callback);
}
@Override
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
index 979e8f460444..6f9315ce2ea1 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
@@ -50,6 +50,10 @@ import org.apache.pulsar.metadata.api.Stat;
public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
private final String sourceMLName;
private volatile Stat sourceLedgersStat;
+ // Kept so that closing this instance unwatches the source only if no other instance has started watching it
+ private final MetaStore.UpdateCallback<ManagedLedgerInfo> sourceManagedLedgerInfoCallback =
+ (managedLedgerInfo, stat) ->
+ executor.execute(() -> processSourceManagedLedgerInfo(managedLedgerInfo, stat));
public ShadowManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper,
MetaStore store, ManagedLedgerConfig config,
@@ -75,10 +79,11 @@ public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
}
private void doInitialize(ManagedLedgerInitializeLedgerCallback callback, Object ctx) {
+ if (failInitializationIfAbandoned(callback)) {
+ return;
+ }
// Fetch the list of existing ledgers in the source managed ledger
- store.watchManagedLedgerInfo(sourceMLName, (managedLedgerInfo, stat) ->
- executor.execute(() -> processSourceManagedLedgerInfo(managedLedgerInfo, stat))
- );
+ store.watchManagedLedgerInfo(sourceMLName, sourceManagedLedgerInfoCallback);
store.getManagedLedgerInfo(sourceMLName, false, null, new MetaStore.MetaStoreCallback<>() {
@Override
public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
@@ -187,6 +192,9 @@ public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
@Override
protected synchronized void initializeBookKeeper(ManagedLedgerInitializeLedgerCallback callback) {
+ if (failInitializationIfAbandoned(callback)) {
+ return;
+ }
log.debug().attr("name", name).attr("ledgers", ledgers).log("Initializing bookkeeper for shadowManagedLedger");
// Calculate total entries and size
@@ -409,7 +417,7 @@ public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
@Override
public synchronized void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx) {
- store.unwatchManagedLedgerInfo(sourceMLName);
+ store.unwatchManagedLedgerInfo(sourceMLName, sourceManagedLedgerInfoCallback);
super.asyncClose(callback, ctx);
}
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 db6e94d5c19c..2857c70c8b25 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
@@ -35,6 +35,7 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
import org.apache.bookkeeper.client.BKException;
@@ -51,6 +52,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.ManagedLedgerInitializeLedgerCallback;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.awaitility.Awaitility;
import org.testng.Assert;
@@ -298,6 +300,164 @@ public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
assertThat(factory.ledgers).isEmpty();
}
+ @Test(timeOut = 30000)
+ public void testTimedOutInitializationIsStoppedAndClosed() throws Exception {
+ CompletableFuture<Void> initializeGate = new CompletableFuture<>();
+ CompletableFuture<ManagedLedgerImpl> gatedLedger = replaceFactoryWithGatedInitialize(initializeGate, null);
+ ManagedLedgerConfig config = defaultConfig().setMetadataOperationsTimeoutSeconds(1);
+ CompletableFuture<ManagedLedger> timedOutOpen = openAsync("ml", config);
+ ManagedLedgerImpl timedOutLedger = gatedLedger.get(5, TimeUnit.SECONDS);
+ // Let the pending initialization time out
+ Thread.sleep(1100);
+
+ // The retried open removes the pending initialization and fails the open waiting for it
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) openAsync("ml", config).get(5, TimeUnit.SECONDS);
+ assertThatThrownBy(() -> timedOutOpen.get(5, TimeUnit.SECONDS))
+ .hasMessageContaining("initialization timed out");
+ assertThat(ledger).isNotSameAs(timedOutLedger);
+ assertThat(ledger.entryCache).isNotSameAs(timedOutLedger.entryCache);
+
+ initializeGate.complete(null);
+ Awaitility.await().untilAsserted(() ->
+ assertThat(timedOutLedger.getState()).isEqualTo(ManagedLedgerImpl.State.Closed));
+ assertThat(factory.ledgers.get("ml").join()).isSameAs(ledger);
+ assertThat(factory.getEntryCacheManager().getEntryCache(ledger)).isSameAs(ledger.entryCache);
+ // The abandoned initialization stopped without opening or creating ledgers, or updating the metadata
+ assertThat(timedOutLedger.getLedgersInfo()).isEmpty();
+ assertThat(metadataStore.get("/managed-ledgers/ml").join().orElseThrow().getStat().getVersion())
+ .isEqualTo(ledger.ledgersStat.getVersion());
+ }
+
+ @Test(timeOut = 30000)
+ public void testOpenRetriesFailedOpen() throws Exception {
+ CompletableFuture<ManagedLedgerImpl> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new ManagedLedgerException("open failed"));
+ factory.ledgers.put("ml", failedFuture);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("ml");
+ assertThat(factory.ledgers.get("ml").join()).isSameAs(ledger);
+ }
+
+ @Test(timeOut = 30000)
+ public void testOpenDoesNotTimeOutInitializationOfAnotherFuture() throws Exception {
+ CompletableFuture<Void> initializeGate = new CompletableFuture<>();
+ CompletableFuture<ManagedLedgerImpl> gatedLedger = replaceFactoryWithGatedInitialize(initializeGate, null);
+ ManagedLedgerConfig config = defaultConfig().setMetadataOperationsTimeoutSeconds(1);
+ CompletableFuture<ManagedLedger> open = openAsync("ml", config);
+ ManagedLedgerImpl pendingLedger = gatedLedger.get(5, TimeUnit.SECONDS);
+ // Let the pending initialization time out
+ Thread.sleep(1100);
+
+ // Simulate an open that read an older future, which a timed out open has replaced since then
+ CompletableFuture<ManagedLedgerImpl> pendingLedgerFuture = factory.ledgers.get("ml");
+ CompletableFuture<ManagedLedgerImpl> olderFuture = new CompletableFuture<>();
+ factory.ledgers.put("ml", olderFuture);
+ CompletableFuture<ManagedLedger> staleOpen = openAsync("ml", config);
+ assertThat(factory.ledgers.get("ml")).isSameAs(olderFuture);
+ factory.ledgers.put("ml", pendingLedgerFuture);
+ olderFuture.completeExceptionally(new ManagedLedgerException("older open failed"));
+ assertThatThrownBy(() -> staleOpen.get(5, TimeUnit.SECONDS)).hasMessageContaining("older open failed");
+
+ // The pending initialization wasn't timed out by the stale open
+ initializeGate.complete(null);
+ assertThat(open.get(5, TimeUnit.SECONDS)).isSameAs(pendingLedger);
+ assertThat(pendingLedger.getState()).isEqualTo(ManagedLedgerImpl.State.LedgerOpened);
+ }
+
+ @Test(timeOut = 30000)
+ public void testTimedOutInitializationFailureKeepsEntryCacheOfRetriedOpen() throws Exception {
+ CompletableFuture<Void> initializeGate = new CompletableFuture<>();
+ CompletableFuture<ManagedLedgerImpl> gatedLedger = replaceFactoryWithGatedInitialize(initializeGate,
+ callback -> callback.initializeFailed(new ManagedLedgerException("injected failure")));
+ ManagedLedgerConfig config = defaultConfig().setMetadataOperationsTimeoutSeconds(1);
+ CompletableFuture<ManagedLedger> timedOutOpen = openAsync("ml", config);
+ ManagedLedgerImpl timedOutLedger = gatedLedger.get(5, TimeUnit.SECONDS);
+ // Let the pending initialization time out
+ Thread.sleep(1100);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) openAsync("ml", config).get(5, TimeUnit.SECONDS);
+
+ initializeGate.complete(null);
+ Awaitility.await().untilAsserted(() ->
+ assertThat(timedOutLedger.getState()).isEqualTo(ManagedLedgerImpl.State.Closed));
+ assertThatThrownBy(() -> timedOutOpen.get(5, TimeUnit.SECONDS))
+ .hasMessageContaining("initialization timed out");
+ assertThat(factory.ledgers.get("ml").join()).isSameAs(ledger);
+ assertThat(factory.getEntryCacheManager().getEntryCache(ledger)).isSameAs(ledger.entryCache);
+ }
+
+ @Test(timeOut = 30000)
+ public void testShutdownClosesInitializationThatCompletesAfterIt() throws Exception {
+ CompletableFuture<Void> initializeGate = new CompletableFuture<>();
+ // The initialization completes without the scheduler of the factory, which the shutdown stops
+ CompletableFuture<ManagedLedgerImpl> gatedLedger = replaceFactoryWithGatedInitialize(initializeGate,
+ ManagedLedgerInitializeLedgerCallback::initializeComplete);
+ factory.setShutdownLedgerCloseTimeoutMillis(100);
+ CompletableFuture<ManagedLedger> open = openAsync("ml", defaultConfig());
+ ManagedLedgerImpl pendingLedger = gatedLedger.get(5, TimeUnit.SECONDS);
+
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ assertThatThrownBy(() -> open.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerException.ManagedLedgerFactoryClosedException.class);
+
+ initializeGate.complete(null);
+ Awaitility.await().untilAsserted(() ->
+ assertThat(pendingLedger.getState()).isEqualTo(ManagedLedgerImpl.State.Closed));
+ }
+
+ private CompletableFuture<ManagedLedger> openAsync(String name, ManagedLedgerConfig config) {
+ CompletableFuture<ManagedLedger> future = new CompletableFuture<>();
+ factory.asyncOpen(name, config, new OpenLedgerCallback() {
+ @Override
+ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
+ future.complete(ledger);
+ }
+
+ @Override
+ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null, null);
+ return future;
+ }
+
+ /**
+ * Replaces the factory with one whose first managed ledger instance only initializes once {@code initializeGate}
+ * completes, by calling {@code initialize} with the initialization callback if it's set. The returned future
+ * completes with the instance when its initialization is requested.
+ */
+ private CompletableFuture<ManagedLedgerImpl> replaceFactoryWithGatedInitialize(
+ CompletableFuture<Void> initializeGate, Consumer<ManagedLedgerInitializeLedgerCallback> initialize)
+ throws Exception {
+ factory.shutdownAsync().get(5, TimeUnit.SECONDS);
+ CompletableFuture<ManagedLedgerImpl> gatedLedger = new CompletableFuture<>();
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc) {
+ @Override
+ protected ManagedLedgerImpl createManagedLedger(BookKeeper bk, MetaStore store, String name,
+ ManagedLedgerConfig config,
+ Supplier<CompletableFuture<Boolean>> mlOwnershipChecker) {
+ if (gatedLedger.isDone()) {
+ return super.createManagedLedger(bk, store, name, config, mlOwnershipChecker);
+ }
+ ManagedLedgerImpl ledger = new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name,
+ mlOwnershipChecker) {
+ @Override
+ synchronized void initialize(ManagedLedgerInitializeLedgerCallback callback, Object ctx) {
+ initializeGate.thenRun(() -> {
+ if (initialize != null) {
+ initialize.accept(callback);
+ } else {
+ super.initialize(callback, ctx);
+ }
+ });
+ gatedLedger.complete(this);
+ }
+ };
+ return ledger;
+ }
+ };
+ return gatedLedger;
+ }
+
/**
* Replaces the factory with one whose managed ledger {@code gatedName} only closes once {@code closeGate}
* completes; {@code closeRequested} completes when its close is requested.
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplTest.java
index 8cc8b6a85570..d214b125195e 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplTest.java
@@ -30,6 +30,7 @@ import static org.testng.Assert.fail;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import lombok.AllArgsConstructor;
@@ -282,6 +283,22 @@ public class MetaStoreImplTest extends MockedBookKeeperTestCase {
promise.get();
}
+ @Test(timeOut = 20000)
+ void unwatchKeepsWatchOfAnotherCallback() throws Exception {
+ MetaStore store = new MetaStoreImpl(metadataStore, executor);
+ metadataStore.put("/managed-ledgers/my_test", "".getBytes(), Optional.empty()).join();
+ CompletableFuture<ManagedLedgerInfo> notified = new CompletableFuture<>();
+ MetaStore.UpdateCallback<ManagedLedgerInfo> replacedCallback = (mlInfo, stat) -> { };
+ store.watchManagedLedgerInfo("my_test", replacedCallback);
+ store.watchManagedLedgerInfo("my_test", (mlInfo, stat) -> notified.complete(mlInfo));
+
+ // Unwatching with the replaced callback, as an instance closed after another one started watching would
+ store.unwatchManagedLedgerInfo("my_test", replacedCallback);
+ metadataStore.put("/managed-ledgers/my_test", "".getBytes(), Optional.empty()).join();
+
+ assertNotNull(notified.get(10, TimeUnit.SECONDS));
+ }
+
@Test
public void testGetChildrenWatch() throws Exception {
MetadataCache<MyClass> objCache1 = metadataStore.getMetadataCache(MyClass.class);