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/12 10:29:08 UTC
(pulsar) branch lh-fix-fenced-ledger-fail-pending created (now 647f1cbd3df8)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch lh-fix-fenced-ledger-fail-pending
in repository https://gitbox.apache.org/repos/asf/pulsar.git
at 647f1cbd3df8 [fix][broker] Stop serving bundles that another broker took over while the metadata session is lost
This branch includes the following new commits:
new 647f1cbd3df8 [fix][broker] Stop serving bundles that another broker took over while the metadata session is lost
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(pulsar) 01/01: [fix][broker] Stop serving bundles that another broker took over while the metadata session is lost
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-fix-fenced-ledger-fail-pending
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 647f1cbd3df8f784a0face98027d3452f0abc633
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Mon Oct 12 13:28:56 2026 +0300
[fix][broker] Stop serving bundles that another broker took over while the metadata session is lost
A broker that has lost its metadata session keeps serving its topics, as the default
zookeeperSessionExpiredPolicy=reconnect intends. When another broker takes a bundle over meanwhile
and fences the topics' ledgers, the isolated broker held its producers' sends until its session
came back, since handling the fenced ledger, the lookups it answered and its attempts to reacquire
the bundle all waited for the metadata store.
- A managed ledger whose ledger another client fenced after the metadata session was lost is
fenced at once, and its pending adds fail with ManagedLedgerFencedByAnotherWriterException; a
recheck of the fenced ledger that waits for the metadata service is replaced by that fencing.
- The topic reports the takeover, and the ownership cache stops serving the bundle and closes its
topics without releasing the lock, for that ownership generation and that session loss only.
- While the session is lost, the lookups and acquisitions that would wait for the metadata store
fail with MetadataSessionLostException, so the clients look the topics up elsewhere, and topic
loads check the ownership again right before opening the managed ledger.
Fixes #26865
Assisted-by: Claude Code (claude-opus-5-5)
---
.../bookkeeper/mledger/ManagedLedgerException.java | 10 ++
.../mledger/impl/ManagedLedgerFactoryImpl.java | 21 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 162 +++++++++++++++++----
.../mledger/impl/ManagedLedgerErrorsTest.java | 98 +++++++++++++
.../org/apache/pulsar/broker/PulsarService.java | 23 +++
.../pulsar/broker/lookup/TopicLookupBase.java | 8 +
.../pulsar/broker/namespace/NamespaceService.java | 46 +++++-
.../pulsar/broker/namespace/OwnershipCache.java | 159 +++++++++++++++++++-
.../pulsar/broker/service/BrokerService.java | 162 +++++++++++----------
.../broker/service/BrokerServiceException.java | 10 ++
.../broker/service/TopicLoadingTracePoints.java | 1 +
.../broker/service/persistent/PersistentTopic.java | 17 +++
.../broker/namespace/OwnershipCacheTest.java | 149 +++++++++++++++++++
.../buffer/TopicTransactionBufferTest.java | 3 +
14 files changed, 750 insertions(+), 119 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
index 1fa565d6ec78..b61c2e8be648 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
@@ -89,6 +89,16 @@ public class ManagedLedgerException extends Exception {
}
}
+ /**
+ * The managed ledger was fenced because another client fenced its current ledger after the metadata store's
+ * session was lost, such as another broker that took the topic over once the session expired.
+ */
+ public static class ManagedLedgerFencedByAnotherWriterException extends ManagedLedgerFencedException {
+ public ManagedLedgerFencedByAnotherWriterException(String message) {
+ super(message);
+ }
+ }
+
public static class ManagedLedgerNotFoundException extends ManagedLedgerException {
public ManagedLedgerNotFoundException(Exception e) {
super(e);
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 cbc8fd9c2bf3..708ec7943706 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
@@ -154,6 +154,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
*/
@Getter
private boolean metadataServiceAvailable;
+ private volatile boolean metadataSessionLost;
private final ManagedLedgerConfig defaultManagedLedgerConfig;
private static class PendingInitializeManagedLedger {
@@ -293,6 +294,26 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
private synchronized void handleMetadataStoreNotification(SessionEvent e) {
log.info().attr("event", e).log("Received MetadataStore session event");
metadataServiceAvailable = e.isConnected();
+ if (e == SessionEvent.SessionLost) {
+ metadataSessionLost = true;
+ // Off the store's event thread and this lock: a managed ledger's recheck of a fenced ledger may be
+ // waiting for the metadata service
+ scheduledExecutor.execute(() -> ledgers.values().forEach(future -> {
+ if (future.isDone() && !future.isCompletedExceptionally()) {
+ future.join().metadataSessionLost();
+ }
+ }));
+ } else if (e == SessionEvent.SessionReestablished) {
+ metadataSessionLost = false;
+ }
+ }
+
+ /**
+ * Whether the metadata store's session was lost and not reestablished yet: the store may have deleted the
+ * session's ephemeral nodes, such as its locks, and other clients may have taken over what they protected.
+ */
+ boolean isMetadataSessionLost() {
+ return metadataSessionLost;
}
private synchronized void flushCursors() {
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 98394275779d..d582381eeae6 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
@@ -111,6 +111,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.CursorNotFoundExcept
import org.apache.bookkeeper.mledger.ManagedLedgerException.InvalidCursorPositionException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.LedgerNotExistException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlreadyClosedException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedByAnotherWriterException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerInterceptException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerNotFoundException;
@@ -289,6 +290,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// ledger metadata stored. This variable used to record the result of the latest once re-checking.
private Pair<Long, CompletableFuture<Integer>> ledgerRecheckInProgress = new ImmutablePair<>(-1L,
CompletableFuture.completedFuture(Code.OK));
+ // The ledger of the recheck in progress and the error that started it, guarded by this: a recheck of a fenced
+ // ledger that started while the metadata session was alive waits for the metadata service, so the managed ledger
+ // is fenced at once if the session is lost meanwhile
+ private LedgerHandle ledgerRecheckHandle;
+ private int ledgerRecheckErrorCode;
public enum State {
None, // Uninitialized
@@ -2075,41 +2081,28 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
synchronized void addEntryFailedDueToConcurrentlyModified(final LedgerHandle currentLedger, int errorCode) {
+ boolean fencedAfterSessionLoss = errorCode == BKException.Code.LedgerFencedException
+ && factory.isMetadataSessionLost();
// The method "addEntryFailedDueToConcurrentlyModified" will be triggered for a certain ledger at most once.
if (ledgerRecheckInProgress.getLeft() == currentLedger.getId()) {
+ // A recheck that started while the metadata session was alive waits for the metadata service; a fencing
+ // after the session was lost replaces it
+ if (fencedAfterSessionLoss && !ledgerRecheckInProgress.getRight().isDone()) {
+ fenceByAnotherWriter(currentLedger, errorCode);
+ }
+ return;
+ }
+ if (fencedAfterSessionLoss) {
+ fenceByAnotherWriter(currentLedger, errorCode);
return;
}
- ledgerRecheckInProgress = new ImmutablePair<>(currentLedger.getId(), new CompletableFuture<>());
+ CompletableFuture<Integer> recheck = new CompletableFuture<>();
+ ledgerRecheckInProgress = new ImmutablePair<>(currentLedger.getId(), recheck);
+ ledgerRecheckHandle = currentLedger;
+ ledgerRecheckErrorCode = errorCode;
OpenCallback opencb = (rc, lh, ctx) -> {
- ledgerRecheckInProgress.getRight().complete(rc);
- if (rc == Code.OK) {
- log.info().attr("ledgerId", lh.getId())
- .attr("inMemoryLac", currentLedger.getLastAddConfirmed())
- .attr("storedLac", lh.getLastAddConfirmed())
- .log("Successfully opened ledger to check the last add confirmed"
- + " position when the ledger was concurrently modified"
- + " (the ledger may be closed by auto-replication)");
- ledgerClosed(currentLedger, lh.getLastAddConfirmed());
- // Close the abandoned write handle, or it leaks with its periodic explicit-LAC flush task.
- currentLedger.asyncClose((closeRc, closedLedger, closeCtx) -> {
- if (closeRc != Code.OK) {
- log.debug().attr("ledgerId", currentLedger.getId())
- .attr("status", BKException.getMessage(closeRc))
- .log("Error when closing ledger after it was concurrently modified");
- }
- }, null);
- } else {
- log.error().attr("ledgerId", currentLedger.getId())
- .attr("lastAddConfirmed", currentLedger.getLastAddConfirmed())
- .attr("rc", rc)
- .log("Fencing the topic to ensure durability and consistency"
- + " (the current ledger was concurrently modified"
- + " by another bookie client, which is not expected)");
- // Stop switching ledger and write topic metadata, to avoid messages lost. The doc of
- // LedgerHandle also mentioned this: https://github.com/apache/bookkeeper/blob/release-4.17.2/
- // bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java#L2047-L2048.
- handleBadVersion(new BadVersionException("the current ledger " + currentLedger.getId()
- + " was concurrent modified by a other bookie client. The error code is: " + errorCode));
+ synchronized (ManagedLedgerImpl.this) {
+ recheckCompleted(currentLedger, recheck, rc, lh, errorCode);
}
};
bookKeeper.newOpenLedgerOp()
@@ -2124,6 +2117,108 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
.whenComplete((rh, ex) -> completeOpenCallback(log, currentLedger.getId(), opencb, rh, ex));
}
+ /**
+ * Another client fenced the current ledger after this broker lost its metadata session: usually a broker that
+ * took the topic over, once the metadata store expired this broker's session and its locks. Without the metadata
+ * service, this broker can neither read the ledger's metadata to check it nor create a new ledger, so the pending
+ * adds would wait until the metadata service returns. Fence the managed ledger instead: the pending adds fail now,
+ * the topic closes, and the clients look the topic up again.
+ */
+ private synchronized void fenceByAnotherWriter(LedgerHandle currentLedger, int errorCode) {
+ ledgerRecheckInProgress = new ImmutablePair<>(currentLedger.getId(),
+ CompletableFuture.completedFuture(errorCode));
+ ledgerRecheckHandle = null;
+ log.warn().attr("ledgerId", currentLedger.getId())
+ .attr("lastAddConfirmed", currentLedger.getLastAddConfirmed())
+ .attr("rc", errorCode)
+ .log("Fencing the managed ledger: another client fenced the current ledger after the metadata"
+ + " session was lost");
+ setFenced(new ManagedLedgerFencedByAnotherWriterException("ManagedLedger " + name + " is fenced: another"
+ + " client fenced its current ledger " + currentLedger.getId() + " after the metadata session was"
+ + " lost"));
+ closeConcurrentlyModifiedLedger(currentLedger);
+ }
+
+ /**
+ * The metadata session was lost: a recheck of a fenced ledger that is still waiting for the metadata service
+ * fences the managed ledger at once, since no other add may fail to tell it.
+ */
+ synchronized void metadataSessionLost() {
+ LedgerHandle recheckHandle = ledgerRecheckHandle;
+ // the session may have been reestablished since the factory scheduled this
+ if (factory.isMetadataSessionLost() && recheckHandle != null
+ && ledgerRecheckErrorCode == BKException.Code.LedgerFencedException
+ && ledgerRecheckInProgress.getLeft() == recheckHandle.getId()
+ && !ledgerRecheckInProgress.getRight().isDone()) {
+ fenceByAnotherWriter(recheckHandle, ledgerRecheckErrorCode);
+ }
+ }
+
+ /** Whether a recheck of the current ledger is waiting for its open with recovery. */
+ @VisibleForTesting
+ synchronized boolean isLedgerRecheckPending() {
+ return ledgerRecheckHandle != null && !ledgerRecheckInProgress.getRight().isDone();
+ }
+
+ private void recheckCompleted(LedgerHandle currentLedger, CompletableFuture<Integer> recheck, int rc,
+ LedgerHandle lh, int errorCode) {
+ recheck.complete(rc);
+ if (lh != null) {
+ // A read handle, opened only to recover the ledger and read its last add confirmed
+ closeRecheckHandle(lh);
+ }
+ if (ledgerRecheckInProgress.getRight() != recheck) {
+ // The managed ledger was fenced meanwhile, after the metadata session was lost
+ log.info().attr("ledgerId", currentLedger.getId()).attr("rc", rc)
+ .log("Ignoring the recheck of a ledger that another client fenced after the metadata session"
+ + " was lost");
+ return;
+ }
+ ledgerRecheckHandle = null;
+ if (rc == Code.OK) {
+ log.info().attr("ledgerId", lh.getId())
+ .attr("inMemoryLac", currentLedger.getLastAddConfirmed())
+ .attr("storedLac", lh.getLastAddConfirmed())
+ .log("Successfully opened ledger to check the last add confirmed"
+ + " position when the ledger was concurrently modified"
+ + " (the ledger may be closed by auto-replication)");
+ ledgerClosed(currentLedger, lh.getLastAddConfirmed());
+ closeConcurrentlyModifiedLedger(currentLedger);
+ } else {
+ log.error().attr("ledgerId", currentLedger.getId())
+ .attr("lastAddConfirmed", currentLedger.getLastAddConfirmed())
+ .attr("rc", rc)
+ .log("Fencing the topic to ensure durability and consistency"
+ + " (the current ledger was concurrently modified"
+ + " by another bookie client, which is not expected)");
+ // Stop switching ledger and write topic metadata, to avoid messages lost. The doc of
+ // LedgerHandle also mentioned this: https://github.com/apache/bookkeeper/blob/release-4.17.2/
+ // bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java#L2047-L2048.
+ handleBadVersion(new BadVersionException("the current ledger " + currentLedger.getId()
+ + " was concurrent modified by a other bookie client. The error code is: " + errorCode));
+ }
+ }
+
+ private void closeRecheckHandle(LedgerHandle lh) {
+ lh.closeAsync().whenComplete((ignore, e) -> {
+ if (e != null) {
+ log.debug().attr("ledgerId", lh.getId()).exceptionMessage(e)
+ .log("Error when closing the handle of a ledger's recheck");
+ }
+ });
+ }
+
+ /** Closes the abandoned write handle, or it leaks with its periodic explicit-LAC flush task. */
+ private void closeConcurrentlyModifiedLedger(LedgerHandle currentLedger) {
+ currentLedger.asyncClose((closeRc, closedLedger, closeCtx) -> {
+ if (closeRc != Code.OK) {
+ log.debug().attr("ledgerId", currentLedger.getId())
+ .attr("status", BKException.getMessage(closeRc))
+ .log("Error when closing ledger after it was concurrently modified");
+ }
+ }, null);
+ }
+
synchronized void ledgerClosed(final LedgerHandle lh) {
ledgerClosed(lh, null);
}
@@ -4786,11 +4881,14 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@VisibleForTesting
public synchronized void setFenced() {
+ setFenced(new ManagedLedgerFencedException("ManagedLedger " + name + " is fenced"));
+ }
+
+ private synchronized void setFenced(ManagedLedgerFencedException pendingAddsException) {
log.info().log("Moving to Fenced state");
State prev = STATE_UPDATER.getAndSet(this, State.Fenced);
if (prev != State.Fenced) {
- clearPendingAddEntries(new ManagedLedgerFencedException("ManagedLedger "
- + name + " is fenced"));
+ clearPendingAddEntries(pendingAddsException);
}
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
index f898fc47298c..23e6a0dcfa9a 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerErrorsTest.java
@@ -21,13 +21,17 @@ package org.apache.bookkeeper.mledger.impl;
import static org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.defaultConfig;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.expectThrows;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
+import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
@@ -48,11 +52,13 @@ import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedByAnotherWriterException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
import org.awaitility.Awaitility;
import org.testng.annotations.Test;
@@ -661,6 +667,98 @@ public class ManagedLedgerErrorsTest extends MockedBookKeeperTestCase {
entries.forEach(Entry::release);
}
+ @Test
+ public void fencedLedgerFencesTheManagedLedgerAfterTheMetadataSessionIsLost() throws Exception {
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory = spy(new ManagedLedgerFactoryImpl(metadataStore, bkc));
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "fencedLedgerFencesTheManagedLedgerAfterTheMetadataSessionIsLost", defaultConfig());
+ ledger.addEntry("entry-1".getBytes());
+ LedgerHandle fencedLedger = ledger.currentLedger;
+
+ // Another broker took the topic over and fenced the ledger after this broker lost its metadata session: the
+ // add must fail at once instead of waiting for the metadata service to check the ledger.
+ when(factory.isMetadataServiceAvailable()).thenReturn(false);
+ when(factory.isMetadataSessionLost()).thenReturn(true);
+ bkc.addEntryFailAfter(0, BKException.Code.LedgerFencedException);
+
+ ManagedLedgerException e = expectThrows(ManagedLedgerException.class,
+ () -> ledger.addEntry("entry-2".getBytes()));
+ assertThat(e, instanceOf(ManagedLedgerFencedByAnotherWriterException.class));
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.Fenced);
+ assertThat(expectThrows(ManagedLedgerException.class, () -> ledger.addEntry("entry-3".getBytes())),
+ instanceOf(ManagedLedgerFencedException.class));
+
+ PulsarMockLedgerHandle mockFencedLedger = (PulsarMockLedgerHandle) fencedLedger;
+ Awaitility.await().untilAsserted(() -> assertTrue(mockFencedLedger.getAsyncCloseCount().get() > 0,
+ "the handle of the concurrently modified ledger must be closed"));
+ }
+
+ @Test
+ public void pendingRecheckOfAFencedLedgerFencesTheManagedLedgerWhenTheMetadataSessionIsLost() throws Exception {
+ // A real factory, not a spy: it learns of the session from the metadata store's events
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "pendingRecheckOfAFencedLedgerFencesTheManagedLedgerWhenTheMetadataSessionIsLost", defaultConfig());
+ ledger.addEntry("entry-1".getBytes());
+ long fencedLedgerId = ledger.currentLedger.getId();
+
+ // The ledger is fenced while only the connection is lost: the recheck opens it with recovery, which waits
+ // for the metadata service
+ metadataStore.triggerSessionEvent(SessionEvent.ConnectionLost);
+ CompletableFuture<Void> recovery = bkc.promiseAfter(0);
+ bkc.addEntryFailAfter(0, BKException.Code.LedgerFencedException);
+ CompletableFuture<Position> add = new CompletableFuture<>();
+ ledger.asyncAddEntry("entry-2".getBytes(), new AddEntryCallback() {
+ @Override
+ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
+ add.complete(position);
+ }
+
+ @Override
+ public void addFailed(ManagedLedgerException exception, Object ctx) {
+ add.completeExceptionally(exception);
+ }
+ }, null);
+ Awaitility.await().until(ledger::isLedgerRecheckPending);
+
+ // The session is lost while the recheck waits: no other add may fail to tell the managed ledger, so the
+ // factory tells it
+ metadataStore.triggerSessionEvent(SessionEvent.SessionLost);
+
+ ExecutionException e = expectThrows(ExecutionException.class, () -> add.get(10, TimeUnit.SECONDS));
+ assertThat(e.getCause(), instanceOf(ManagedLedgerFencedByAnotherWriterException.class));
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.Fenced);
+
+ // The recheck completing later doesn't switch the fenced managed ledger to a new ledger
+ recovery.complete(null);
+ Awaitility.await().pollDelay(Duration.ofMillis(500)).untilAsserted(() -> {
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.Fenced);
+ assertEquals(ledger.currentLedger.getId(), fencedLedgerId);
+ });
+ }
+
+ @Test
+ public void fencedLedgerIsReplacedWhileTheMetadataSessionIsAlive() throws Exception {
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory = spy(new ManagedLedgerFactoryImpl(metadataStore, bkc));
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
+ "fencedLedgerIsReplacedWhileTheMetadataSessionIsAlive", defaultConfig());
+ ledger.addEntry("entry-1".getBytes());
+ long fencedLedgerId = ledger.currentLedger.getId();
+
+ // Only the connection is lost, so the session and its locks are still this broker's, and no other broker
+ // can own the topic: the ledger, fenced by someone else such as auto-recovery, is replaced as before.
+ when(factory.isMetadataServiceAvailable()).thenReturn(false);
+ when(factory.isMetadataSessionLost()).thenReturn(false);
+ bkc.addEntryFailAfter(0, BKException.Code.LedgerFencedException);
+
+ ledger.addEntry("entry-2".getBytes());
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.LedgerOpened);
+ assertNotEquals(ledger.currentLedger.getId(), fencedLedgerId);
+ }
+
@Test
public void recoverAfterMarkDeleteError() throws Exception {
ManagedLedger ledger = factory.open("my_test_ledger", initManagedLedgerConfig(defaultConfig()));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 5a4486b71663..3c1bd4151c72 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -63,6 +63,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@@ -333,6 +334,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
Init, Started, Closing, Closed
}
+ private volatile boolean metadataSessionLost;
+ // How many times the local metadata store's session was lost, to tell one loss from the next
+ private final AtomicLong metadataSessionLosses = new AtomicLong();
private volatile State state;
private final ReentrantLock mutex = new ReentrantLock();
@@ -1280,8 +1284,27 @@ public class PulsarService implements AutoCloseable, ShutdownService {
addBrokerAdditionalServlets(webService, attributeMap, config);
}
+ /**
+ * Whether the local metadata store's session was lost and not reestablished yet. The store may have deleted the
+ * session's ephemeral nodes meanwhile, and another broker may have acquired the bundles whose locks they were.
+ */
+ public boolean isMetadataSessionLost() {
+ return metadataSessionLost;
+ }
+
+ /** How many times the local metadata store's session was lost since the broker started. */
+ public long getMetadataSessionLossCount() {
+ return metadataSessionLosses.get();
+ }
+
private void handleMetadataSessionEvent(SessionEvent e) {
log.info().attr("event", e).log("Received metadata service session event");
+ if (e == SessionEvent.SessionLost) {
+ metadataSessionLosses.incrementAndGet();
+ metadataSessionLost = true;
+ } else if (e == SessionEvent.SessionReestablished) {
+ metadataSessionLost = false;
+ }
if (e == SessionEvent.SessionLost
&& config.getZookeeperSessionExpiredPolicy() == MetadataSessionExpiredPolicy.shutdown) {
log.warn()
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
index faa066db6e3d..23574b41f4a5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
@@ -36,6 +36,7 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.namespace.LookupOptions;
+import org.apache.pulsar.broker.service.BrokerServiceException.MetadataSessionLostException;
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse.LookupType;
@@ -359,6 +360,13 @@ public class TopicLookupBase extends PulsarWebResource {
.attr("errorMessage", errorMsg)
.log("Failed to lookup topic");
lookupFuture.complete(newLookupErrorResponse(ServerError.MetadataError, errorMsg, requestId));
+ } else if (unwrapEx instanceof MetadataSessionLostException) {
+ // the client looks the topic up again, so it's expected for every lookup while the session is lost
+ LOG.debug()
+ .attr("topic", topicName)
+ .attr("errorMessage", errorMsg)
+ .log("Failed to lookup topic");
+ lookupFuture.complete(newLookupErrorResponse(ServerError.ServiceNotReady, errorMsg, requestId));
} else if (unwrapEx instanceof MetadataStoreException) {
// Load bundle ownership or acquire lock failed.
// Differ with "IllegalStateException", print warning LOG.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 1f95c281d493..ccc40ac80f70 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -45,6 +45,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -68,6 +69,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl
import org.apache.pulsar.broker.loadbalance.extensions.manager.RedirectManagerForLoadManagerMigration;
import org.apache.pulsar.broker.lookup.LookupResult;
import org.apache.pulsar.broker.resources.NamespaceResources;
+import org.apache.pulsar.broker.service.BrokerServiceException.MetadataSessionLostException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
@@ -285,6 +287,23 @@ public class NamespaceService implements AutoCloseable {
return bundles.map(b -> b.findBundle(topicName));
}
+ /**
+ * Stops serving the bundle of a topic whose ledger another broker fenced after this broker lost its metadata
+ * session, which shows that the other broker took the bundle over. See
+ * {@link OwnershipCache#handleOwnershipTakenOver(TopicName)}. The extensible load manager tracks the ownerships
+ * in its own service unit state channel, which this doesn't apply to.
+ */
+ public void handleOwnershipTakenOver(TopicName topicName) {
+ if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+ return;
+ }
+ ownershipCache.handleOwnershipTakenOver(topicName).exceptionally(e -> {
+ log.warn().attr("topic", topicName).exception(e)
+ .log("Failed to stop serving the bundle that another broker took over");
+ return null;
+ });
+ }
+
public NamespaceBundle getBundle(TopicName topicName) {
return bundleFactory.getBundles(topicName.getNamespaceObject()).findBundle(topicName);
}
@@ -483,7 +502,9 @@ public class NamespaceService implements AutoCloseable {
}
LookupRequestKey lookupRequestKey = LookupRequestKey.from(bundle, options);
- return targetMap.computeIfAbsent(lookupRequestKey, (k) -> {
+ AtomicBoolean started = new AtomicBoolean();
+ CompletableFuture<Optional<LookupResult>> lookup = targetMap.computeIfAbsent(lookupRequestKey, (k) -> {
+ started.set(true);
CompletableFuture<Optional<LookupResult>> future = new CompletableFuture<>();
// First check if we or someone else already owns the bundle
@@ -507,10 +528,16 @@ public class NamespaceService implements AutoCloseable {
resolveBrokerServiceLookupResult(options, nsData.get(), future);
}
}).exceptionally(exception -> {
- log.warn()
- .attr("bundle", bundle)
- .exception(exception)
- .log("Failed to check owner for bundle");
+ if (FutureUtil.unwrapCompletionException(exception) instanceof MetadataSessionLostException) {
+ // the client retries the lookup
+ log.debug().attr("bundle", bundle).exceptionMessage(exception)
+ .log("Failed to check owner for bundle");
+ } else {
+ log.warn()
+ .attr("bundle", bundle)
+ .exception(exception)
+ .log("Failed to check owner for bundle");
+ }
future.completeExceptionally(exception);
return null;
});
@@ -521,6 +548,15 @@ public class NamespaceService implements AutoCloseable {
return future;
});
+ if (!started.get() && !lookup.isDone() && pulsar.isMetadataSessionLost()) {
+ // Another lookup of the bundle is still in progress while the session is lost: it may be waiting for the
+ // metadata store, such as for acquiring the bundle, which this can't tell from a lookup that is only
+ // choosing a broker to redirect to. Fail rather than risk waiting behind it: the client looks the topic
+ // up again, possibly at another broker.
+ return FutureUtil.failedFuture(new MetadataSessionLostException("Namespace bundle " + bundle
+ + " is being looked up while this broker's metadata session is lost"));
+ }
+ return lookup;
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
index a45a17e2e931..ecd5bafed343 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
@@ -34,12 +34,16 @@ import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import lombok.CustomLog;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.BrokerServiceException.MetadataSessionLostException;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundles;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.stats.CacheMetricsCollector;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.coordination.LockManager;
@@ -227,6 +231,10 @@ public class OwnershipCache {
*/
public CompletableFuture<Optional<NamespaceEphemeralData>> getOwnerAsync(NamespaceBundle suName) {
CompletableFuture<OwnedBundle> ownedBundleFuture = ownedBundlesCache.getIfPresent(suName);
+ if (ownedBundleFuture != null && !ownedBundleFuture.isDone() && pulsar.isMetadataSessionLost()) {
+ // an acquisition that started before the session was lost waits for the metadata store
+ return FutureUtil.failedFuture(sessionLostException(suName));
+ }
if (ownedBundleFuture != null) {
// Either we're the owners or we're trying to become the owner.
return ownedBundleFuture.thenApply(serviceUnit -> {
@@ -240,6 +248,12 @@ public class OwnershipCache {
return lockManager.readLock(path).thenCompose(owner -> {
// If the current broker is the owner, attempt to reacquire ownership to avoid cache loss.
if (owner.isPresent() && owner.get().equals(selfOwnerInfo)) {
+ if (pulsar.isMetadataSessionLost()) {
+ // The metadata cache can still hold this broker's lock value although the store deleted it with
+ // the session, and another broker may own the bundle now. Reacquiring it would wait for the
+ // metadata store, so fail at once, and the client looks the topic up at another broker.
+ return FutureUtil.failedFuture(sessionLostException(suName));
+ }
log.warn()
.attr("broker", selfOwnerInfo)
.attr("bundle", suName)
@@ -260,6 +274,11 @@ public class OwnershipCache {
});
}
+ private static MetadataSessionLostException sessionLostException(NamespaceBundle bundle) {
+ return new MetadataSessionLostException("Namespace bundle " + bundle
+ + " can't be acquired by this broker while its metadata session is lost");
+ }
+
/**
* Method to get the current owner of the <code>NamespaceBundle</code>
* or set the local broker as the owner if absent.
@@ -270,6 +289,10 @@ public class OwnershipCache {
* @throws Exception
*/
public CompletableFuture<NamespaceEphemeralData> tryAcquiringOwnership(NamespaceBundle bundle) throws Exception {
+ if (pulsar.isMetadataSessionLost()) {
+ // acquiring the lock would wait for the session to be reestablished
+ return FutureUtil.failedFuture(sessionLostException(bundle));
+ }
if (!refreshSelfOwnerInfo()) {
return FutureUtil.failedFuture(
new RuntimeException("Namespace service is not ready for acquiring ownership"));
@@ -280,12 +303,19 @@ public class OwnershipCache {
// Doing a get() on the ownedBundlesCache will trigger an async metadata write to acquire the lock over the
// service unit. Serialized against both removeOwnership(NamespaceBundle) and removeOwnership(OwnedBundle)
// for the same bundle: see bundleOperationBarriers.
- return serialize(bundle, () -> ownedBundlesCache.get(bundle)
- .thenApply(namespaceBundle -> {
- log.info().attr("bundle", namespaceBundle).log("Successfully acquired ownership");
- namespaceService.onNamespaceBundleOwned(bundle);
- return selfOwnerInfo;
- }));
+ return serialize(bundle, () -> {
+ // Again when the acquisition's turn comes: an acquisition queued behind another operation of the bundle
+ // may start after the session was lost
+ if (pulsar.isMetadataSessionLost()) {
+ return FutureUtil.failedFuture(sessionLostException(bundle));
+ }
+ return ownedBundlesCache.get(bundle)
+ .thenApply(namespaceBundle -> {
+ log.info().attr("bundle", namespaceBundle).log("Successfully acquired ownership");
+ namespaceService.onNamespaceBundleOwned(bundle);
+ return selfOwnerInfo;
+ });
+ });
}
/**
@@ -505,6 +535,123 @@ public class OwnershipCache {
this.ownedBundlesCache.synchronous().invalidate(namespaceBundle);
}
+ /**
+ * Stops serving the bundle of a topic whose ledger another broker fenced after this broker lost its metadata
+ * session, which shows that the other broker took the bundle over.
+ *
+ * <p>A broker that has lost its metadata session keeps serving its bundles from this cache until it can
+ * revalidate their locks. When the metadata store expires the session, it deletes the session's locks, and
+ * another broker can acquire a bundle's lock and load its topics, fencing the topics' ledgers. This broker's
+ * writes then fail: that is the only sign of the takeover it gets before its session is reestablished. Without
+ * acting on it, this broker keeps answering lookups for the bundle with itself and keeps loading its topics,
+ * which wait for the metadata store, so the clients can't move to the new owner until then.
+ *
+ * <p>This removes the bundle's ownership generation from the cache and closes the bundle's topics, as an unload
+ * does, but doesn't release the lock, since releasing it needs the metadata store and another broker holds it.
+ * While the session is lost, this broker doesn't acquire bundles (see {@link #tryAcquiringOwnership}) nor
+ * reacquire a bundle whose lock value the metadata cache still holds as its own (see
+ * {@link #getOwnerAsync(NamespaceBundle)}), and topic loads fail the ownership check, so the clients look the
+ * topic up again until they reach the new owner. When the session is reestablished, the lock manager
+ * revalidates the lock: it finds another broker's lock and expires it, and its expiry listener finds this
+ * generation gone and has nothing left to do. If no other broker holds the lock after all, the lock revalidates,
+ * and the next lookup reacquires the ownership from it with a new lock instance, as after any loss of the cache
+ * entry; the revalidated instance is then no longer tracked and needs nothing.
+ *
+ * <p>The handling runs on the broker's executor in the bundle's operation order with the acquisitions (see
+ * {@link #serialize}), until the closing of the topics has started, and only while the session is lost, when no
+ * acquisition can complete, so it can't race with an acquisition. A report that comes after the session was
+ * reestablished is ignored, leaving the takeover to the lock's revalidation. The report applies to the ownership
+ * generation that served the topic when it was made and to that loss of the session only, so a report that runs
+ * after the bundle was acquired again, or after the session came back and was lost again, changes nothing.
+ * Besides, only the loaded instance of a topic reports (see {@code PersistentTopic.addFailed}).
+ *
+ * @return a future that completes when the report has been handled: the bundle is no longer served and its
+ * topics' closing has started, or the report was ignored
+ */
+ CompletableFuture<Void> handleOwnershipTakenOver(TopicName topicName) {
+ // Find the bundle among the owned ones, which needs no metadata, rather than in the namespace's bundles,
+ // which can have been invalidated in the cache.
+ OwnedBundle reported = null;
+ for (CompletableFuture<OwnedBundle> future : ownedBundlesCache.asMap().values()) {
+ OwnedBundle candidate = future.isDone() && !future.isCompletedExceptionally() ? future.join() : null;
+ if (candidate != null && candidate.getNamespaceBundle().includes(topicName)) {
+ reported = candidate;
+ break;
+ }
+ }
+ if (reported == null || !pulsar.isMetadataSessionLost()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ // The report applies to this ownership generation and this loss of the session only: a report that runs
+ // after the bundle was reacquired, or after the session came back and was lost again, changes nothing
+ OwnedBundle reportedBundle = reported;
+ long sessionLoss = pulsar.getMetadataSessionLossCount();
+ NamespaceBundle takenOverBundle = reported.getNamespaceBundle();
+ // On the broker's executor, not on the thread that reported the takeover, and the bundle's next operation
+ // waits until the closing of its topics has started, so that the closing can't reach a newer generation's
+ return serialize(takenOverBundle, () -> {
+ try {
+ return CompletableFuture.runAsync(() -> stopServingTakenOverBundle(takenOverBundle, reportedBundle,
+ sessionLoss, topicName), pulsar.getExecutor());
+ } catch (RejectedExecutionException e) {
+ // the broker is shutting down
+ stopServingTakenOverBundle(takenOverBundle, reportedBundle, sessionLoss, topicName);
+ return CompletableFuture.completedFuture(null);
+ }
+ });
+ }
+
+ private void stopServingTakenOverBundle(NamespaceBundle bundle, OwnedBundle ownedBundle, long sessionLoss,
+ TopicName topicName) {
+ if (!pulsar.isMetadataSessionLost() || pulsar.getMetadataSessionLossCount() != sessionLoss) {
+ return;
+ }
+ if (getOwnedBundle(bundle) != ownedBundle) {
+ // another topic of the bundle reported the takeover first, or the bundle was acquired again since
+ return;
+ }
+ AtomicBoolean invalidated = new AtomicBoolean();
+ // the entry is a completed future, so the invalidation runs synchronously
+ invalidateLocalOwnerCache(bundle, ownedBundle, () -> invalidated.set(true));
+ if (!invalidated.get()) {
+ return;
+ }
+ ownedBundle.setActive(false);
+ log.warn().attr("bundle", bundle).attr("topic", topicName)
+ .log("Another broker took the bundle over after the metadata session was lost; stopped serving it"
+ + " and closing its topics");
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures;
+ try {
+ topicFutures = pulsar.getBrokerService().getTopicFuturesInBundle(bundle);
+ } catch (RuntimeException e) {
+ log.warn().attr("bundle", bundle).exception(e)
+ .log("Failed to close the topics of a bundle that another broker took over");
+ namespaceService.onNamespaceBundleUnload(bundle);
+ return;
+ }
+ closeTakenOverTopics(bundle, topicFutures);
+ }
+
+ private void closeTakenOverTopics(NamespaceBundle bundle,
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures) {
+ CompletableFuture<Integer> unloadFuture;
+ try {
+ unloadFuture = pulsar.getBrokerService().unloadServiceUnit(bundle, true, true,
+ pulsar.getConfiguration().getNamespaceBundleUnloadingTimeoutMs(), TimeUnit.MILLISECONDS,
+ topicFutures);
+ } catch (RuntimeException e) {
+ unloadFuture = FutureUtil.failedFuture(e);
+ }
+ unloadFuture.whenComplete((unloadedTopics, ex) -> {
+ if (ex != null) {
+ log.warn().attr("bundle", bundle).exception(ex)
+ .log("Failed to close the topics of a bundle that another broker took over");
+ }
+ pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle, topicFutures);
+ namespaceService.onNamespaceBundleUnload(bundle);
+ });
+ }
+
/**
* Invalidate the local owner cache entry once it holds the given {@link OwnedBundle} instance, so that a
* stale lock-expiry callback cannot drop an entry installed by a newer acquisition. {@code onInvalidated}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 045de9cf9d67..1629362f88d6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2314,96 +2314,106 @@ public class BrokerService implements Closeable {
loggerContextBuilder.attr("topic", topicName.toString());
}
managedLedgerConfig.setLoggerContext(loggerContextBuilder.build());
- final LatencyTracer.TracePoint openMlTracePoint = context.startTrace(TopicLoadingTracePoints.OPEN_ML);
- managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig,
- new OpenLedgerCallback() {
- @Override
- public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
- try {
- context.finishTrace(openMlTracePoint, null);
- PersistentTopic persistentTopic = isSystemTopic(topic)
- ? new SystemTopic(topic, ledger, BrokerService.this)
- : newTopic(topic, ledger, BrokerService.this, PersistentTopic.class);
- persistentTopic.setCreateFuture(topicFuture);
- context.trace(TopicLoadingTracePoints.INIT, persistentTopic.initialize(context))
- .thenCompose(__ -> context.trace(
- TopicLoadingTracePoints.PRE_CREATE_COMPACTED_SUB,
- persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
- .thenCompose(__ -> context.trace(TopicLoadingTracePoints.REPLICATION,
- persistentTopic.initializeCheckReplication()))
- .thenCompose(v -> context.trace(TopicLoadingTracePoints.DEDUPLICATION,
- persistentTopic.checkDeduplicationStatus()))
- .thenRun(() -> {
- final var latency = context.getSnapshot();
- if (!topicFuture.complete(Optional.of(persistentTopic))) {
- // Check create persistent topic timeout.
- if (topicFuture.isCompletedExceptionally()) {
- log.warn()
- .attr("topic", topic)
- .attr("error",
- FutureUtil.getException(topicFuture).orElse(null))
- .attr("latency", latency.description())
- .log("The future is already completed with failure, "
- + "closing the topic");
+ // Check the ownership again right before opening the managed ledger: the steps above can wait for the
+ // metadata store, and a broker that stopped serving the bundle meanwhile, such as one that another broker
+ // took the bundle over from, must not open, and so recover, the new owner's ledger.
+ context.trace(TopicLoadingTracePoints.OWNERSHIP_BEFORE_OPEN_ML, checkTopicNsOwnership(topic))
+ .thenRun(() -> {
+ final LatencyTracer.TracePoint openMlTracePoint = context.startTrace(TopicLoadingTracePoints.OPEN_ML);
+ managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig,
+ new OpenLedgerCallback() {
+ @Override
+ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
+ try {
+ context.finishTrace(openMlTracePoint, null);
+ PersistentTopic persistentTopic = isSystemTopic(topic)
+ ? new SystemTopic(topic, ledger, BrokerService.this)
+ : newTopic(topic, ledger, BrokerService.this, PersistentTopic.class);
+ persistentTopic.setCreateFuture(topicFuture);
+ context.trace(TopicLoadingTracePoints.INIT, persistentTopic.initialize(context))
+ .thenCompose(__ -> context.trace(
+ TopicLoadingTracePoints.PRE_CREATE_COMPACTED_SUB,
+ persistentTopic.preCreateSubscriptionForCompactionIfNeeded()))
+ .thenCompose(__ -> context.trace(TopicLoadingTracePoints.REPLICATION,
+ persistentTopic.initializeCheckReplication()))
+ .thenCompose(v -> context.trace(TopicLoadingTracePoints.DEDUPLICATION,
+ persistentTopic.checkDeduplicationStatus()))
+ .thenRun(() -> {
+ final var latency = context.getSnapshot();
+ if (!topicFuture.complete(Optional.of(persistentTopic))) {
+ // Check create persistent topic timeout.
+ if (topicFuture.isCompletedExceptionally()) {
+ log.warn()
+ .attr("topic", topic)
+ .attr("error", FutureUtil.getException(
+ topicFuture).orElse(null))
+ .attr("latency", latency.description())
+ .log("The future is already completed with failure, "
+ + "closing the topic");
+ } else {
+ // It should not happen.
+ log.error()
+ .attr("topic", topic)
+ .log("future is already completed by another"
+ + " thread, which is not expected. Closing"
+ + " the current one");
+ }
+ executor().submit(() -> {
+ persistentTopic.close().whenComplete((ignore, ex) -> {
+ topics.remove(topic, topicFuture);
+ if (ex != null) {
+ log.warn()
+ .attr("topic", topic)
+ .exception(ex)
+ .log("Get an error when closing topic.");
+ }
+ });
+ });
} else {
- // It should not happen.
- log.error()
- .attr("topic", topic)
- .log("future is already completed by another thread, "
- + "which is not expected. Closing the current one");
+ addTopicToStatsMaps(topicName, persistentTopic);
}
+ })
+ .exceptionally((ex) -> {
+ log.warn()
+ .attr("topic", topic)
+ .exceptionMessage(ex)
+ .log("Topic initialization failed. Removing topic from cache");
executor().submit(() -> {
- persistentTopic.close().whenComplete((ignore, ex) -> {
+ persistentTopic.close().whenComplete((ignore, closeEx) -> {
topics.remove(topic, topicFuture);
- if (ex != null) {
+ if (closeEx != null) {
log.warn()
.attr("topic", topic)
- .exception(ex)
.log("Get an error when closing topic.");
}
+ topicFuture.completeExceptionally(ex);
});
});
- } else {
- addTopicToStatsMaps(topicName, persistentTopic);
- }
- })
- .exceptionally((ex) -> {
- log.warn()
- .attr("topic", topic)
- .exceptionMessage(ex)
- .log("Topic initialization failed. Removing topic from cache");
- executor().submit(() -> {
- persistentTopic.close().whenComplete((ignore, closeEx) -> {
- topics.remove(topic, topicFuture);
- if (closeEx != null) {
- log.warn()
- .attr("topic", topic)
- .log("Get an error when closing topic.");
- }
- topicFuture.completeExceptionally(ex);
- });
+ return null;
});
- return null;
- });
- } catch (Exception e) {
- failTopicFuture(topic, topicFuture, e);
+ } catch (Exception e) {
+ failTopicFuture(topic, topicFuture, e);
+ }
}
- }
- @Override
- public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
- if (!createIfMissing && exception instanceof ManagedLedgerNotFoundException) {
- context.finishTrace(openMlTracePoint, null);
- // We were just trying to load a topic and the topic doesn't exist
- pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
- loadFuture.completeExceptionally(exception);
- topicFuture.complete(Optional.empty());
- } else {
- context.finishTrace(openMlTracePoint, exception);
- failTopicFuture(topic, topicFuture, new PersistenceException(exception));
+ @Override
+ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ if (!createIfMissing && exception instanceof ManagedLedgerNotFoundException) {
+ context.finishTrace(openMlTracePoint, null);
+ // We were just trying to load a topic and the topic doesn't exist
+ pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+ loadFuture.completeExceptionally(exception);
+ topicFuture.complete(Optional.empty());
+ } else {
+ context.finishTrace(openMlTracePoint, exception);
+ failTopicFuture(topic, topicFuture, new PersistenceException(exception));
+ }
}
- }
- }, () -> isTopicNsOwnedByBrokerAsync(topicName), null);
+ }, () -> isTopicNsOwnedByBrokerAsync(topicName), null);
+ }).exceptionally(ex -> {
+ failTopicFuture(topic, topicFuture, FutureUtil.unwrapCompletionException(ex));
+ return null;
+ });
return null;
}).exceptionally((exception) -> {
boolean migrationFailure = exception.getCause() instanceof TopicMigratedException;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
index 95c5550d3b21..5ca7f867e485 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
@@ -71,6 +71,16 @@ public class BrokerServiceException extends Exception {
}
}
+ /**
+ * A service unit can't be acquired, or looked up without waiting for the metadata store, while the broker's
+ * metadata session is lost. The client looks the topic up again, possibly at another broker.
+ */
+ public static class MetadataSessionLostException extends ServiceUnitNotReadyException {
+ public MetadataSessionLostException(String msg) {
+ super(msg);
+ }
+ }
+
public static class TopicClosedException extends BrokerServiceException {
public TopicClosedException(Throwable t) {
super(t);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingTracePoints.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingTracePoints.java
index bfefe62ef2a6..232158b9f81c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingTracePoints.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicLoadingTracePoints.java
@@ -35,6 +35,7 @@ public final class TopicLoadingTracePoints {
public static final String CHECK_TOPIC_ALREADY_MIGRATED = "check-topic-already-migrated";
public static final String VALIDATE_TOPIC_CONSISTENCY = "validate-topic-consistency";
public static final String ML_CONFIG = "ml-config";
+ public static final String OWNERSHIP_BEFORE_OPEN_ML = "ownership-before-open-ml";
public static final String OPEN_ML = "open-ml";
public static final String INIT = "init";
public static final String PRE_CREATE_COMPACTED_SUB = "pre-create-compacted-sub";
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 81ba81476cd4..77f478f0cc8b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -79,6 +79,7 @@ import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlreadyClosedException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedByAnotherWriterException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerTerminatedException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.MetadataNotFoundException;
@@ -218,6 +219,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private static final Logger LOG = Logger.get(PersistentTopic.class);
protected final Logger log;
+ // Whether a write showed that another broker took the topic over, which is reported once
+ private final AtomicBoolean ownershipTakeoverReported = new AtomicBoolean();
+
// Managed ledger associated with the topic
protected final ManagedLedger ledger;
@@ -818,6 +822,19 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
if (exception instanceof ManagedLedgerFencedException) {
+ if (exception instanceof ManagedLedgerFencedByAnotherWriterException
+ && brokerService.getTopicReference(topic).orElse(null) == this
+ && ownershipTakeoverReported.compareAndSet(false, true)) {
+ // Only the loaded instance of the topic reports: an instance that an unload already replaced, whose
+ // write failure is reported late, belongs to an older ownership generation of the bundle.
+ // Another broker fenced the ledger after this broker lost its metadata session: it took the topic's
+ // bundle over, so stop serving the bundle instead of answering lookups for it with this broker
+ try {
+ brokerService.getPulsar().getNamespaceService().handleOwnershipTakenOver(TopicName.get(topic));
+ } catch (RuntimeException e) {
+ log.warn().exception(e).log("Failed to stop serving the bundle that another broker took over");
+ }
+ }
// If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen
close();
callback.completed(new TopicFencedException(exception.getMessage()), -1, -1);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
index f9fef81287b5..fe915d0f8690 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/OwnershipCacheTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -53,16 +54,20 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.IntStream;
import lombok.Cleanup;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.logging.log4j.Level;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
@@ -81,6 +86,8 @@ import org.testng.annotations.Test;
@Test(groups = "broker")
public class OwnershipCacheTest {
+ private final AtomicBoolean metadataSessionLost = new AtomicBoolean();
+ private final AtomicLong metadataSessionLossCount = new AtomicLong();
private PulsarService pulsar;
private ServiceConfiguration config;
@@ -100,6 +107,11 @@ public class OwnershipCacheTest {
selfBrokerUrl = "tcp://localhost:" + port;
pulsar = mock(PulsarService.class);
when(pulsar.isRunning()).thenReturn(true);
+ // Answered from fields that the tests set, rather than stubbed again in a test while other threads call them
+ metadataSessionLost.set(false);
+ metadataSessionLossCount.set(0);
+ doAnswer(inv -> metadataSessionLost.get()).when(pulsar).isMetadataSessionLost();
+ doAnswer(inv -> metadataSessionLossCount.get()).when(pulsar).getMetadataSessionLossCount();
config = new ServiceConfiguration();
executor = OrderedScheduler.newSchedulerBuilder().numThreads(1).name("test").build();
zookeeperServer = new ZookeeperServerTest(0);
@@ -370,6 +382,143 @@ public class OwnershipCacheTest {
assertEquals(cache.getOwnedBundles().size(), 1);
}
+ @Test
+ public void testHandleOwnershipTakenOver() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-taken-over"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+ // topics whose hashes fall in the bundle, which covers half of the namespace's hash range
+ List<TopicName> topics = IntStream.range(0, 100)
+ .mapToObj(i -> TopicName.get("persistent://pulsar/ns-taken-over/topic-" + i))
+ .filter(bundle::includes).limit(2).toList();
+ assertEquals(topics.size(), 2);
+ cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle ownedBundle = cache.getOwnedBundle(bundle);
+ assertNotNull(ownedBundle);
+ ResourceLock<NamespaceEphemeralData> lock = cache.getLocallyAcquiredLocks().get(bundle);
+
+ // a report while the session is alive is left to the lock's revalidation
+ cache.handleOwnershipTakenOver(topics.get(0)).get();
+ assertSame(cache.getOwnedBundle(bundle), ownedBundle);
+
+ metadataSessionLost.set(true);
+ cache.handleOwnershipTakenOver(topics.get(0)).get();
+
+ // The broker stopped serving the bundle: topic loads fail the ownership check, and the bundle's topics are
+ // closed.
+ assertNull(cache.getOwnedBundle(bundle));
+ assertFalse(ownedBundle.isActive());
+ assertFalse(cache.checkOwnershipAsync(bundle).get());
+ Awaitility.await().untilAsserted(() -> {
+ verify(brokerService).unloadServiceUnit(same(bundle), eq(true), eq(true), anyLong(), any(), any());
+ verify(nsService).onNamespaceBundleUnload(bundle);
+ });
+ // The lock isn't released, since releasing it needs the metadata store: its revalidation after the session
+ // is reestablished expires it.
+ assertSame(cache.getLocallyAcquiredLocks().get(bundle), lock);
+ assertTrue(store.exists(ServiceUnitUtils.path(bundle)).join());
+
+ // another topic of the bundle reporting the takeover changes nothing
+ cache.handleOwnershipTakenOver(topics.get(1)).get();
+ verify(brokerService, times(1)).unloadServiceUnit(same(bundle), eq(true), eq(true), anyLong(), any(), any());
+ verify(nsService, times(1)).onNamespaceBundleUnload(bundle);
+
+ // While the session is lost, a lookup that finds this broker's own lock value doesn't reacquire the bundle,
+ // which would wait for the metadata store: it fails, and the client looks the topic up elsewhere.
+ assertThatFailsWithServiceUnitNotReady(cache.getOwnerAsync(bundle));
+ assertNull(cache.getOwnedBundle(bundle));
+
+ // When the session is back and no other broker took the lock after all, the lookup reacquires the bundle.
+ metadataSessionLost.set(false);
+ Optional<NamespaceEphemeralData> owner = cache.getOwnerAsync(bundle).get();
+ assertTrue(owner.isPresent());
+ assertEquals(owner.get().getNativeUrl(), selfBrokerUrl);
+ assertNotNull(cache.getOwnedBundle(bundle));
+ assertTrue(cache.checkOwnershipAsync(bundle).get());
+ }
+
+ @Test
+ public void testNoAcquisitionWhileTheMetadataSessionIsLost() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle owned = new NamespaceBundle(NamespaceName.get("pulsar/ns-session-lost"),
+ Range.closedOpen(0L, 0x40000000L), bundleFactory);
+ NamespaceBundle notOwned = new NamespaceBundle(NamespaceName.get("pulsar/ns-session-lost"),
+ Range.closedOpen(0x40000000L, 0x80000000L), bundleFactory);
+ cache.tryAcquiringOwnership(owned).get();
+
+ metadataSessionLost.set(true);
+
+ // the bundle it serves is still served, as with the reconnect policy for an expired session
+ assertTrue(cache.getOwnerAsync(owned).get().isPresent());
+ assertTrue(cache.checkOwnershipAsync(owned).get());
+ // another bundle is looked up as before, but not acquired
+ assertFalse(cache.getOwnerAsync(notOwned).get().isPresent());
+ assertThatFailsWithServiceUnitNotReady(cache.tryAcquiringOwnership(notOwned));
+ assertNull(cache.getOwnedBundle(notOwned));
+
+ metadataSessionLost.set(false);
+ assertEquals(cache.tryAcquiringOwnership(notOwned).get().getNativeUrl(), selfBrokerUrl);
+ }
+
+ @Test
+ public void testQueuedAcquisitionFailsWhenTheMetadataSessionIsLostMeanwhile() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-queued-acquisition"),
+ Range.closedOpen(0L, 0x40000000L), bundleFactory);
+ // another operation of the bundle is in progress, so the acquisition waits for its turn
+ CompletableFuture<Void> preceding = new CompletableFuture<>();
+ cache.getBundleOperationBarriers().put(bundle, preceding);
+ CompletableFuture<NamespaceEphemeralData> acquisition = cache.tryAcquiringOwnership(bundle);
+
+ // the session is lost before its turn comes: acquiring the lock would wait for the metadata store
+ metadataSessionLost.set(true);
+ preceding.complete(null);
+
+ assertThatFailsWithServiceUnitNotReady(acquisition);
+ assertNull(cache.getOwnedBundle(bundle));
+ assertFalse(store.exists(ServiceUnitUtils.path(bundle)).join());
+ }
+
+ @Test
+ public void testTakeoverReportOfAnEarlierSessionLossChangesNothing() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-earlier-loss"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE), bundleFactory);
+ TopicName topic = IntStream.range(0, 100)
+ .mapToObj(i -> TopicName.get("persistent://pulsar/ns-earlier-loss/topic-" + i))
+ .filter(bundle::includes).findFirst().orElseThrow();
+ cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle ownedBundle = cache.getOwnedBundle(bundle);
+
+ // the report of the first loss of the session waits behind another operation of the bundle
+ metadataSessionLost.set(true);
+ metadataSessionLossCount.set(1);
+ CompletableFuture<Void> preceding = new CompletableFuture<>();
+ cache.getBundleOperationBarriers().put(bundle, preceding);
+ CompletableFuture<Void> report = cache.handleOwnershipTakenOver(topic);
+
+ // meanwhile the session came back, the lock revalidated, and the session was lost again
+ metadataSessionLossCount.set(2);
+ preceding.complete(null);
+ report.get(5, TimeUnit.SECONDS);
+
+ assertSame(cache.getOwnedBundle(bundle), ownedBundle);
+ assertTrue(ownedBundle.isActive());
+ verify(brokerService, never()).unloadServiceUnit(same(bundle), anyBoolean(), anyBoolean(), anyLong(), any(),
+ any());
+ }
+
+ private static void assertThatFailsWithServiceUnitNotReady(CompletableFuture<?> future) {
+ try {
+ future.get(5, TimeUnit.SECONDS);
+ fail("the future should have failed");
+ } catch (Exception e) {
+ assertTrue(e instanceof ExecutionException
+ && e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException, e.toString());
+ }
+ }
+
@Test
public void testRemoveOwnership() throws Exception {
OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TopicTransactionBufferTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TopicTransactionBufferTest.java
index e0583b5aaf7d..de2022d9a781 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TopicTransactionBufferTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TopicTransactionBufferTest.java
@@ -180,6 +180,9 @@ public class TopicTransactionBufferTest extends TransactionTestBase {
.newTopic(Mockito.eq(topic), Mockito.any(), Mockito.eq(brokerService),
Mockito.eq(PersistentTopic.class));
+ // createPersistentTopic0 runs after the ownership check of the topic's load, which this test skips: its
+ // check right before opening the managed ledger finds the topic's bundle owned
+ Mockito.doReturn(CompletableFuture.completedFuture(null)).when(brokerService).checkTopicNsOwnership(topic);
brokerService.createPersistentTopic0(new TopicLoadingContext(TopicName.get(topic), true,
new CompletableFuture<>(), Mockito.mock(PulsarStats.class)));