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/08/04 15:02:32 UTC
(pulsar) 06/08: [fix][ml] Tolerate concurrent creation of the managed ledger z-node (#26247)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cd889206c1be9e50c2238c699fbdce529014251c
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jul 27 10:17:48 2026 +0300
[fix][ml] Tolerate concurrent creation of the managed ledger z-node (#26247)
(cherry picked from commit d54d83df25ca3ebdb7574a80f31e5252d5d91995)
---
.../bookkeeper/mledger/impl/MetaStoreImpl.java | 11 +++++
.../bookkeeper/mledger/impl/MetaStoreImplTest.java | 48 ++++++++++++++++++++++
2 files changed, 59 insertions(+)
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 4e7adf9aeaf5..a1d4b109d468 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
@@ -132,6 +132,17 @@ public class MetaStoreImpl implements MetaStore, Consumer<Notification> {
}
callback.operationComplete(ledgerBuilder, stat);
}).exceptionally(ex -> {
+ if (FutureUtil.unwrapCompletionException(ex)
+ instanceof MetadataStoreException.BadVersionException) {
+ // The z-node was created concurrently after the read above returned
+ // "not found". This happens for example when the broker creates the
+ // partitions of a topic while the same topic is being loaded. Read the
+ // z-node back instead of failing the managed ledger initialization.
+ log.info("Managed ledger path '{}' was concurrently created, reading it "
+ + "back", path);
+ getManagedLedgerInfo(ledgerName, false, null, callback);
+ return null;
+ }
callback.operationFailed(getException(ex));
return null;
});
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 e6802906791e..e7a524c30107 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
@@ -18,6 +18,10 @@
*/
package org.apache.bookkeeper.mledger.impl;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
@@ -25,6 +29,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.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -37,6 +42,7 @@ import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.metadata.api.MetadataCache;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
import org.testng.annotations.Test;
@@ -158,6 +164,48 @@ public class MetaStoreImplTest extends MockedBookKeeperTestCase {
promise.get();
}
+ /**
+ * The z-node backing a managed ledger can be created by another party (for example
+ * {@code AdminResource#tryCreatePartitionAsync}) in between the read and the write that
+ * {@code getManagedLedgerInfo(createIfMissing = true)} performs. The metadata store reports that as a
+ * {@link MetadataStoreException.BadVersionException}, which must not fail the managed ledger initialization.
+ */
+ @Test(timeOut = 20000)
+ void createMLNodeConcurrently() throws Exception {
+ String ledgerName = "my_test";
+ String path = "/managed-ledgers/" + ledgerName;
+
+ AtomicBoolean raceInjected = new AtomicBoolean();
+ MetadataStoreExtended racyStore = spy(metadataStore);
+ doAnswer(invocation -> {
+ if (path.equals(invocation.getArgument(0)) && raceInjected.compareAndSet(false, true)) {
+ // Create the z-node behind the caller's back and report it as still missing, so that the
+ // subsequent create hits an already existing z-node.
+ metadataStore.put(path, new byte[0], Optional.of(-1L)).get();
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ return invocation.callRealMethod();
+ }).when(racyStore).get(anyString());
+
+ MetaStore store = new MetaStoreImpl(racyStore, executor);
+
+ final CompletableFuture<ManagedLedgerInfo> promise = new CompletableFuture<>();
+ store.getManagedLedgerInfo(ledgerName, true, new MetaStoreCallback<ManagedLedgerInfo>() {
+ public void operationFailed(MetaStoreException e) {
+ promise.completeExceptionally(e);
+ }
+
+ public void operationComplete(ManagedLedgerInfo result, Stat version) {
+ promise.complete(result);
+ }
+ });
+
+ ManagedLedgerInfo info = promise.get();
+ assertNotNull(info);
+ assertEquals(info.getLedgerInfosCount(), 0);
+ assertTrue(raceInjected.get());
+ }
+
@Test(timeOut = 20000)
void updatingCursorNode() throws Exception {
MetaStore store = new MetaStoreImpl(metadataStore, executor);