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/09/10 12:13:03 UTC
(pulsar) branch master updated: [fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup (#26197)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 37d64fe63055 [fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup (#26197)
37d64fe63055 is described below
commit 37d64fe63055df52f4de7519cd71603a9617fa0a
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Thu Sep 10 20:12:56 2026 +0800
[fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup (#26197)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
---
.../channel/ServiceUnitStateChannelImpl.java | 23 +-
.../pulsar/broker/namespace/NamespaceService.java | 52 +-
.../pulsar/broker/namespace/OwnedBundle.java | 86 +++-
.../pulsar/broker/namespace/OwnershipCache.java | 217 ++++++++-
.../pulsar/broker/service/BrokerService.java | 135 ++++--
.../broker/namespace/NamespaceServiceTest.java | 29 ++
.../broker/namespace/OwnershipCacheTest.java | 530 +++++++++++++++++++++
.../pulsar/broker/service/BrokerServiceTest.java | 87 +++-
8 files changed, 1083 insertions(+), 76 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index 972ed1068935..e38a74c4a938 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -87,6 +87,7 @@ import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
import org.apache.pulsar.broker.namespace.LookupOptions;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Schema;
@@ -1085,12 +1086,28 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
long startTime = System.nanoTime();
MutableInt unloadedTopics = new MutableInt();
NamespaceBundle bundle = LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit);
+ // Capture the topic futures for this bundle once, before unloading starts, and reuse the same snapshot
+ // both to close the topics below and to clean up afterward, so the cleanup is identity-safe: a stale
+ // cleanup call must only remove what this snapshot observed, never a newer ownership generation's topic
+ // installed for the same bundle in the meantime.
+ //
+ // A straggler topic loaded into BrokerService's topic cache for this bundle *after* the snapshot is
+ // taken is deliberately left alone: neither closed below nor evicted by the cleanup. The alternative —
+ // force-evicting whatever is in the cache at cleanup time regardless of identity — could yank a topic
+ // that was never closed, letting a second, independent instance for the same name be created on the
+ // next load. A straggler surviving the unload does not need this unload to fix it: new connections
+ // converge through the normal lookup path once ownership has actually moved, and BookKeeper's ledger
+ // fencing (see ManagedLedgerImpl#addEntryFailedDueToConcurrentlyModified) is what prevents it from
+ // silently double-writing if a new owner's ManagedLedger instance does start writing the same topic.
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures =
+ pulsar.getBrokerService().getTopicFuturesInBundle(bundle);
return pulsar.getBrokerService().unloadServiceUnit(
bundle,
disconnectClients,
true,
pulsar.getConfig().getNamespaceBundleUnloadingTimeoutMs(),
- TimeUnit.MILLISECONDS)
+ TimeUnit.MILLISECONDS,
+ topicFutures)
.thenApply(numUnloadedTopics -> {
unloadedTopics.setValue(numUnloadedTopics);
return numUnloadedTopics;
@@ -1098,7 +1115,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
.whenComplete((__, ex) -> {
if (disconnectClients) {
// clean up topics that failed to unload from the broker ownership cache
- pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
+ pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle, topicFutures);
}
pulsar.getNamespaceService().onNamespaceBundleUnload(bundle);
double unloadBundleTime = TimeUnit.NANOSECONDS
@@ -1108,7 +1125,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
.exception(ex)
.log("Failed to close topics under bundle");
if (!disconnectClients) {
- pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
+ pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle, topicFutures);
}
} else {
log.info().attr("bundle", bundle).attr("topicCount", unloadedTopics)
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 c6d2af76d08d..b538fb772d10 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
@@ -25,11 +25,13 @@ import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.client.api.PulsarClientException.FailedFeatureCheck.SupportsGetPartitionedMetadataWithoutAutoCreation;
import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.hash.Hashing;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.prometheus.client.Counter;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -127,7 +129,10 @@ public class NamespaceService implements AutoCloseable {
private final ServiceConfiguration config;
private final AtomicReference<LoadManager> loadManager;
private final PulsarService pulsar;
- private final OwnershipCache ownershipCache;
+ // Not final so that tests in this package can install a spy through setOwnershipCache(); volatile keeps
+ // the safe-publication guarantee that the final field used to provide, since this is read from the
+ // lookup path on many threads.
+ private volatile OwnershipCache ownershipCache;
private final MetadataCache<LocalBrokerData> localBrokerDataCache;
private final NamespaceBundleFactory bundleFactory;
private final String host;
@@ -1070,12 +1075,44 @@ public class NamespaceService implements AutoCloseable {
// success updateNamespaceBundles
// disable old bundle in memory
getOwnershipCache().updateBundleState(bundle, false)
- .thenRun(() -> {
+ .thenCompose(__ -> {
// update bundled_topic cache for load-report-generation
pulsar.getBrokerService().refreshTopicToStatsMaps(bundle);
loadManager.get().setLoadReportForceUpdateFlag();
- // release old bundle from ownership cache
- pulsar.getNamespaceService().getOwnershipCache().removeOwnership(bundle);
+ // Release old bundle from ownership cache. Compose on the returned future instead of
+ // discarding it, so a delayed or failed release is observed here rather than letting
+ // completionFuture complete while the release may still be in flight; a release
+ // failure is logged and does not fail the split, which has already succeeded.
+ // The wait is bounded: the release is queued behind any in-flight acquire of the same
+ // bundle, and an acquire stuck on an unreachable metadata store has no timeout of its
+ // own, so an unbounded wait here could hold the split's completion hostage. On timeout
+ // the release keeps running in the background and the split completes.
+ CompletableFuture<Void> release =
+ pulsar.getNamespaceService().getOwnershipCache().removeOwnership(bundle);
+ CompletableFuture<Void> boundedRelease = new CompletableFuture<>();
+ release.whenComplete((released, releaseEx) -> {
+ if (releaseEx != null) {
+ boundedRelease.completeExceptionally(releaseEx);
+ } else {
+ boundedRelease.complete(released);
+ }
+ });
+ FutureUtil.addTimeoutHandling(boundedRelease,
+ Duration.ofSeconds(config.getMetadataStoreOperationTimeoutSeconds()),
+ pulsar.getExecutor(),
+ () -> FutureUtil.createTimeoutException(
+ "Timed out waiting for the ownership of the old bundle " + bundle
+ + " to be released after split",
+ NamespaceService.class, "splitAndOwnBundleOnceAndRetry"));
+ return boundedRelease.exceptionally(ex1 -> {
+ log.warn()
+ .attr("bundle", bundle.toString())
+ .exception(ex1)
+ .log("Failed to release ownership of the old bundle after split");
+ return null;
+ });
+ })
+ .thenRun(() -> {
completionFuture.complete(null);
if (unload) {
// Unload new split bundles, in background. This will not
@@ -1087,7 +1124,7 @@ public class NamespaceService implements AutoCloseable {
.exceptionally(e -> {
String msg1 = format(
"failed to disable bundle %s under namespace [%s] with error %s",
- bundle.getNamespaceObject().toString(), bundle, ex.getMessage());
+ bundle.getNamespaceObject().toString(), bundle, e.getMessage());
log.warn().exception(e).log(msg1);
completionFuture.completeExceptionally(new ServiceUnitNotReadyException(msg1));
return null;
@@ -1208,6 +1245,11 @@ public class NamespaceService implements AutoCloseable {
.setLocalPoliciesWithVersion(nsname, localPolicies, nsBundles.getVersion());
}
+ @VisibleForTesting
+ void setOwnershipCache(OwnershipCache ownershipCache) {
+ this.ownershipCache = ownershipCache;
+ }
+
public OwnershipCache getOwnershipCache() {
return ownershipCache;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
index 71fd8efa1aec..f44d4a9c5d59 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.broker.namespace;
+import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -27,8 +29,10 @@ import lombok.CustomLog;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.api.coordination.ResourceLock;
@EqualsAndHashCode
@ToString
@@ -36,6 +40,15 @@ import org.apache.pulsar.common.util.FutureUtil;
public class OwnedBundle {
private final NamespaceBundle bundle;
+ /**
+ * The resource lock acquired for this ownership. Ties this instance to a single ownership generation: a
+ * bundle can be re-acquired (new lock, new {@link OwnedBundle}) after this instance's lock expired, and
+ * cleanup done through this instance must never touch the newer generation.
+ */
+ @ToString.Exclude
+ @EqualsAndHashCode.Exclude
+ private final ResourceLock<NamespaceEphemeralData> resourceLock;
+
/**
* {@link #nsLock} is used to protect read/write access to {@link #isActive} flag and the corresponding code section
* based on {@link #isActive} flag.
@@ -53,23 +66,39 @@ public class OwnedBundle {
* constructor.
*
* @param suName
+ * @deprecated an {@link OwnedBundle} created this way is not bound to a {@link ResourceLock}, so it
+ * represents no ownership generation: {@link OwnershipCache#removeOwnership(OwnedBundle)} releases nothing
+ * for it. Instances that own a bundle are created by {@link OwnershipCache} only.
*/
+ @Deprecated
public OwnedBundle(NamespaceBundle suName) {
- this.bundle = suName;
- IS_ACTIVE_UPDATER.set(this, TRUE);
- };
+ this(suName, true);
+ }
/**
* Constructor to allow set initial active flag.
*
* @param suName
* @param active
+ * @deprecated see {@link #OwnedBundle(NamespaceBundle)}.
*/
+ @Deprecated
public OwnedBundle(NamespaceBundle suName, boolean active) {
this.bundle = suName;
+ this.resourceLock = null;
IS_ACTIVE_UPDATER.set(this, active ? TRUE : FALSE);
}
+ OwnedBundle(NamespaceBundle suName, ResourceLock<NamespaceEphemeralData> resourceLock) {
+ this.bundle = suName;
+ this.resourceLock = resourceLock;
+ IS_ACTIVE_UPDATER.set(this, TRUE);
+ }
+
+ ResourceLock<NamespaceEphemeralData> getResourceLock() {
+ return resourceLock;
+ }
+
/**
* Access to the namespace name.
*
@@ -130,11 +159,48 @@ public class OwnedBundle {
AtomicInteger unloadedTopics = new AtomicInteger();
log.info().attr("ownership", this.bundle).log("Disabling ownership");
+ // Capture the topic futures for this bundle once, before unloading starts, and reuse the same snapshot
+ // both to close the topics below and to clean up afterward: if this generation's lock has already expired
+ // and the bundle is re-acquired while the close futures are still running, the cleanup step must only
+ // touch what this snapshot observed, never a newer generation's topic.
+ //
+ // A straggler topic loaded into BrokerService's topic cache for this bundle *after* the snapshot is
+ // taken is deliberately left alone: neither closed below nor evicted by the cleanup. The alternative —
+ // force-evicting whatever is in the cache at cleanup time regardless of identity — could yank a topic
+ // that was never closed, letting a second, independent instance for the same name be created on the
+ // next load. A straggler surviving the unload does not need this unload to fix it: new connections
+ // converge through the normal lookup path once ownership has actually moved, and BookKeeper's ledger
+ // fencing (see ManagedLedgerImpl#addEntryFailedDueToConcurrentlyModified) is what prevents it from
+ // silently double-writing if a new owner's ManagedLedger instance does start writing the same topic.
+ //
+ // Both the snapshot and the start of the unload are guarded against synchronous exceptions: isActive has
+ // already been flipped above and cannot be flipped back, so letting an exception escape here (instead of
+ // feeding it into the chain below, which treats a topic-close failure as non-fatal and still releases the
+ // ownership) would leave this bundle inactive with its lock still registered, and every retried unload
+ // would fail fast with "Namespace is not active".
+ Map<String, CompletableFuture<Optional<Topic>>> snapshot;
+ try {
+ snapshot = pulsar.getBrokerService().getTopicFuturesInBundle(bundle);
+ } catch (RuntimeException e) {
+ log.warn().attr("bundle", bundle).exception(e).log("Failed to snapshot the topics of the bundle");
+ snapshot = Map.of();
+ }
+ final Map<String, CompletableFuture<Optional<Topic>>> topicFutures = snapshot;
+
// close topics forcefully
- return pulsar.getNamespaceService().getOwnershipCache()
- .updateBundleState(this.bundle, false)
- .thenCompose(v -> pulsar.getBrokerService().unloadServiceUnit(
- bundle, true, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit))
+ // isActive was already flipped to false above; looking the bundle up in the ownership cache here could
+ // deactivate a newer OwnedBundle that re-acquired the bundle after this instance's lock expired.
+ CompletableFuture<Integer> unloadFuture;
+ try {
+ unloadFuture = pulsar.getBrokerService().unloadServiceUnit(
+ bundle, true, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit, topicFutures);
+ } catch (RuntimeException e) {
+ // The topic closes may already have been started from the snapshot before the exception (e.g. the
+ // unload timeout could not be scheduled), so the cleanup below keeps using the same snapshot.
+ unloadFuture = FutureUtil.failedFuture(e);
+ }
+
+ return unloadFuture
.handle((numUnloadedTopics, ex) -> {
if (ex != null) {
// ignore topic-close failure to unload bundle
@@ -146,12 +212,12 @@ public class OwnedBundle {
unloadedTopics.set(numUnloadedTopics);
}
// clean up topics that failed to unload from the broker ownership cache
- pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle);
+ pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle, topicFutures);
return null;
})
.thenCompose(v -> {
- // delete ownership node on zk
- return pulsar.getNamespaceService().getOwnershipCache().removeOwnership(bundle);
+ // delete ownership node on zk, but only for this instance's ownership generation
+ return pulsar.getNamespaceService().getOwnershipCache().removeOwnership(this);
}).whenComplete((ignored, ex) -> {
double unloadBundleTime = TimeUnit.NANOSECONDS
.toMillis((System.nanoTime() - unloadBundleStartTime));
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 bc37d1f99062..b52a587d777f 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
@@ -31,8 +31,11 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
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.AtomicReference;
+import java.util.function.Supplier;
import lombok.CustomLog;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.common.naming.NamespaceBundle;
@@ -73,6 +76,22 @@ public class OwnershipCache {
private final Map<NamespaceBundle, ResourceLock<NamespaceEphemeralData>> locallyAcquiredLocks;
+ /**
+ * Serializes, per bundle, this cache's own acquire ({@link #tryAcquiringOwnership(NamespaceBundle)}) against
+ * both of its release paths: the generation-agnostic {@link #removeOwnership(NamespaceBundle)} and the
+ * generation-aware {@link #removeOwnership(OwnedBundle)} — the one every normal
+ * {@link OwnedBundle#handleUnloadRequest} unload actually goes through. All three touch
+ * {@link #locallyAcquiredLocks} without otherwise coordinating with each other: an acquire installs its lock
+ * asynchronously, well after the cache entry that triggered it becomes visible, while a release removes
+ * whatever lock is currently registered for the bundle (blindly for the {@code NamespaceBundle}-keyed
+ * overload, or only if it still matches a specific generation for the {@code OwnedBundle}-keyed one).
+ * Without this barrier, a release can run in the gap between those two points and either release a lock a
+ * concurrent acquire just installed (reporting success to that acquire's caller while the underlying ZK
+ * lock is gone) or run before the install and be silently bypassed by it (reporting ownership fully
+ * released while a fresh acquisition lands moments later).
+ */
+ private final Map<NamespaceBundle, CompletableFuture<Void>> bundleOperationBarriers = new ConcurrentHashMap<>();
+
/**
* The loading cache of locally owned <code>NamespaceBundle</code> objects.
*/
@@ -92,14 +111,56 @@ public class OwnershipCache {
return lockManager.acquireLock(ServiceUnitUtils.path(namespaceBundle), selfOwnerInfo)
.thenApply(rl -> {
locallyAcquiredLocks.put(namespaceBundle, rl);
+ OwnedBundle ownedBundle = new OwnedBundle(namespaceBundle, rl);
rl.getLockExpiredFuture()
.thenRun(() -> {
log.info().attr("path", rl.getPath()).log("Resource lock has expired");
- namespaceService.unloadNamespaceBundle(namespaceBundle);
- invalidateLocalOwnerCache(namespaceBundle);
- namespaceService.onNamespaceBundleUnload(namespaceBundle);
+ locallyAcquiredLocks.remove(namespaceBundle, rl);
+ // Only unload the generation this listener belongs to.
+ // unloadNamespaceBundle resolves the owner by bundle name, so a listener
+ // that runs late (the load already failed the publication check below and
+ // a lookup re-acquired the bundle in the meantime) would otherwise close a
+ // newer generation's topics and release its lock. If the cache does not
+ // hold this instance, this generation never served anything (or is already
+ // gone), so there is nothing to unload.
+ if (getOwnedBundle(namespaceBundle) == ownedBundle) {
+ namespaceService.unloadNamespaceBundle(namespaceBundle)
+ .exceptionally(ex -> {
+ log.debug()
+ .attr("bundle", namespaceBundle)
+ .exception(ex)
+ .log("Failed to unload namespace bundle after its"
+ + " resource lock expired");
+ return null;
+ });
+ }
+ // The unload event is tied to the same generation check: it is delivered
+ // only if this generation's entry was the one published in the cache (and is
+ // removed here). A generation that never got published never produced an
+ // owned event, and if the cache holds a newer generation that one is still
+ // owned; in both cases an unload event would be unmatched. Listeners keep
+ // per-namespace bookkeeping on these events (the topic-policies service counts
+ // owned bundles and tears the namespace's cache down at zero), so an
+ // unmatched event is not harmless.
+ invalidateLocalOwnerCache(namespaceBundle, ownedBundle,
+ () -> namespaceService.onNamespaceBundleUnload(namespaceBundle));
});
- return new OwnedBundle(namespaceBundle);
+ if (rl.getLockExpiredFuture().isDone()) {
+ // Expiry won the race: never publish an OwnedBundle whose lock is already gone. Let
+ // this load fail instead; Caffeine removes a failed load from the cache automatically,
+ // so no separate cache invalidation is needed here, and the caller of
+ // tryAcquiringOwnership observes a failure instead of a fleeting, already-invalid
+ // success. Reading the future's own state covers both the lock that was already
+ // expired when the listener above was registered (the listener ran inline) and one
+ // that expired concurrently with the registration (the listener may still be
+ // running on the completing thread). A lock that expires right after this check
+ // still converges once the listener runs: see
+ // invalidateLocalOwnerCache(NamespaceBundle, OwnedBundle).
+ throw new IllegalStateException(
+ "Lock for bundle " + namespaceBundle
+ + " expired before ownership could be published");
+ }
+ return ownedBundle;
});
}
}
@@ -209,13 +270,14 @@ public class OwnershipCache {
log.info().attr("bundle", bundle).log("Trying to acquire ownership");
// Doing a get() on the ownedBundlesCache will trigger an async metadata write to acquire the lock over the
- // service unit
- return ownedBundlesCache.get(bundle)
+ // 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;
- });
+ log.info().attr("bundle", namespaceBundle).log("Successfully acquired ownership");
+ namespaceService.onNamespaceBundleOwned(bundle);
+ return selfOwnerInfo;
+ }));
}
/**
@@ -223,13 +285,105 @@ public class OwnershipCache {
*
*/
public CompletableFuture<Void> removeOwnership(NamespaceBundle bundle) {
- ResourceLock<NamespaceEphemeralData> lock = locallyAcquiredLocks.remove(bundle);
+ // Serialized against tryAcquiringOwnership(NamespaceBundle) for the same bundle: see
+ // bundleOperationBarriers. This also subsumes waiting for a concurrently in-flight acquire to settle
+ // before this blind, generation-agnostic release runs.
+ return serialize(bundle, () -> {
+ ResourceLock<NamespaceEphemeralData> lock = locallyAcquiredLocks.remove(bundle);
+ if (lock == null) {
+ // We don't own the specified bundle anymore
+ return CompletableFuture.completedFuture(null);
+ }
+
+ return lock.release();
+ });
+ }
+
+ /**
+ * Runs {@code operation} only after any previously queued {@link #tryAcquiringOwnership(NamespaceBundle)},
+ * {@link #removeOwnership(NamespaceBundle)}, or {@link #removeOwnership(OwnedBundle)} call for the same
+ * bundle has settled, and queues subsequent calls for that bundle behind this one in turn. The barrier entry
+ * for a bundle is removed once no further operation is queued behind it, so {@link #bundleOperationBarriers}
+ * does not grow unboundedly. Each operation's completion hands the barrier to the next queued operation via
+ * {@link PulsarService#getExecutor()}, so the depth of the drain does not depend on the length of the queue
+ * (and a caller's completion callbacks never run the next queued operation on their own stack).
+ *
+ * <p>The {@code compute} call below only captures the preceding barrier and installs the new one; it does not
+ * chain {@code operation} itself. {@code ConcurrentHashMap.compute} runs its remapping function while holding
+ * the map's per-bucket lock, and a same-thread reentrant call into {@code compute} for the same key from
+ * inside that function is a usage the map's contract leaves undefined. If {@code operation} (or the preceding
+ * barrier) were chained inline here, an already-completed source — a fast-failing metadata call, or the
+ * completed futures test doubles commonly return — would make {@code thenCompose} run {@code operation}
+ * synchronously right there, still under that lock; were {@code operation} to then reach {@code serialize()}
+ * again for the same bundle, that would be exactly such a reentrant call. Building the chain after
+ * {@code compute} returns means even a fully synchronous, self-reentrant {@code operation} only ever produces
+ * ordinary, non-nested {@code compute} calls.
+ */
+ private <T> CompletableFuture<T> serialize(NamespaceBundle bundle, Supplier<CompletableFuture<T>> operation) {
+ CompletableFuture<T> result = new CompletableFuture<>();
+ CompletableFuture<Void> opDone = new CompletableFuture<>();
+ AtomicReference<CompletableFuture<Void>> precedingOpRef = new AtomicReference<>();
+ bundleOperationBarriers.compute(bundle, (k, previous) -> {
+ precedingOpRef.set(previous != null ? previous : CompletableFuture.completedFuture(null));
+ return opDone;
+ });
+ precedingOpRef.get().handle((r, e) -> null)
+ .thenCompose(ignore -> operation.get())
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ result.completeExceptionally(e);
+ } else {
+ result.complete(r);
+ }
+ // Release the barrier on the executor rather than inline: the next queued operation runs
+ // as a dependent of opDone, and a cache-hit acquire completes synchronously, so completing
+ // inline would run each queued operation nested inside the previous one's stack frame. A
+ // deep enough backlog then overflows the stack, the StackOverflowError is swallowed by the
+ // CompletableFuture machinery, and this barrier entry is never released, leaving the bundle
+ // permanently unacquirable and unreleasable on this broker.
+ try {
+ opDone.completeAsync(() -> null, pulsar.getExecutor());
+ } catch (RejectedExecutionException ree) {
+ // The executor is gone (broker shutting down): still release the barrier, inline, so
+ // that the operations queued behind it (e.g. the releases of the remaining bundles) can
+ // run instead of hanging forever.
+ opDone.complete(null);
+ }
+ });
+ opDone.whenComplete((r, e) -> bundleOperationBarriers.remove(bundle, opDone));
+ return result;
+ }
+
+ /**
+ * Method to remove the ownership that was acquired for the given {@link OwnedBundle} instance only.
+ *
+ * <p>If the bundle has since been re-acquired (the given instance's lock expired and a newer
+ * {@link OwnedBundle} with a newer lock owns the bundle now), the newer ownership is left untouched. An
+ * instance that is not bound to a lock at all (not created by this cache) releases nothing.
+ *
+ * <p>Serialized against {@link #tryAcquiringOwnership(NamespaceBundle)} for the same bundle: see
+ * {@link #bundleOperationBarriers}. This is the release path every normal {@link OwnedBundle#handleUnloadRequest}
+ * unload actually goes through, so without this barrier a concurrent acquire could still observe and
+ * report success for the generation this call is in the middle of releasing.
+ */
+ public CompletableFuture<Void> removeOwnership(OwnedBundle ownedBundle) {
+ ResourceLock<NamespaceEphemeralData> lock = ownedBundle.getResourceLock();
if (lock == null) {
- // We don't own the specified bundle anymore
+ // The instance is not bound to a lock (not created by this cache), so it represents no ownership
+ // generation and there is nothing of its own to release. Releasing whatever generation currently
+ // owns the bundle instead would be exactly the generation-blind release this method exists to avoid.
+ log.warn().attr("bundle", ownedBundle.getNamespaceBundle())
+ .log("Ignoring release request for an OwnedBundle that is not bound to a resource lock");
return CompletableFuture.completedFuture(null);
}
-
- return lock.release();
+ return serialize(ownedBundle.getNamespaceBundle(), () -> {
+ if (!locallyAcquiredLocks.remove(ownedBundle.getNamespaceBundle(), lock)) {
+ // This ownership generation was already released or has expired; a newer acquisition may own the
+ // bundle now and must not be disturbed.
+ return CompletableFuture.completedFuture(null);
+ }
+ return lock.release();
+ });
}
/**
@@ -343,11 +497,46 @@ public class OwnershipCache {
this.ownedBundlesCache.synchronous().invalidate(namespaceBundle);
}
+ /**
+ * 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}
+ * runs only if this call actually removed that entry, i.e. only for a generation that had been published.
+ *
+ * <p>The callback that calls this can run before the cache's own load future for this bundle has completed:
+ * the lock-expiry listener is registered inside the cache loader's {@code thenApply}, and if the lock was
+ * already expired at that point the listener fires inline, synchronously, before the loader returns and the
+ * future is published as done. (If the lock instead expires concurrently with registration, {@code thenRun}
+ * gives no such synchronous guarantee — the listener may run later, on whichever thread completes the
+ * future — but that only changes when this method is called relative to publication, not whether it is
+ * called; the handling below covers both.) Waiting on the future via {@code whenComplete} instead of
+ * requiring {@code isDone()} up front handles both cases: if the future is already done the callback runs
+ * immediately, and if not, the removal is deferred until the loader publishes it, so the newly-published
+ * {@link OwnedBundle} — whose lock has already expired — is not left claiming active ownership forever.
+ */
+ private void invalidateLocalOwnerCache(NamespaceBundle namespaceBundle, OwnedBundle expectedOwnedBundle,
+ Runnable onInvalidated) {
+ CompletableFuture<OwnedBundle> future = ownedBundlesCache.getIfPresent(namespaceBundle);
+ if (future == null) {
+ return;
+ }
+ future.whenComplete((ownedBundle, ex) -> {
+ if (ex == null && ownedBundle == expectedOwnedBundle
+ && ownedBundlesCache.asMap().remove(namespaceBundle, future)) {
+ onInvalidated.run();
+ }
+ });
+ }
+
@VisibleForTesting
public Map<NamespaceBundle, ResourceLock<NamespaceEphemeralData>> getLocallyAcquiredLocks() {
return locallyAcquiredLocks;
}
+ @VisibleForTesting
+ Map<NamespaceBundle, CompletableFuture<Void>> getBundleOperationBarriers() {
+ return bundleOperationBarriers;
+ }
+
public synchronized boolean refreshSelfOwnerInfo() {
this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getBrokerId(), pulsar.getBrokerServiceUrl(),
pulsar.getBrokerServiceUrlTls(), pulsar.getWebServiceAddress(),
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 1a7375cc6299..01ea66d73956 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
@@ -2882,8 +2882,24 @@ public class BrokerService implements Closeable {
public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
boolean disconnectClients,
boolean closeWithoutWaitingClientDisconnect, long timeout, TimeUnit unit) {
+ return unloadServiceUnit(serviceUnit, disconnectClients, closeWithoutWaitingClientDisconnect, timeout, unit,
+ getTopicFuturesInBundle(serviceUnit));
+ }
+
+ /**
+ * Same as {@link #unloadServiceUnit(NamespaceBundle, boolean, boolean, long, TimeUnit)}, but takes the topic
+ * futures to unload as given, instead of scanning {@link #topics} again. Callers that also need to run
+ * {@link #cleanUnloadedTopicFromCache(NamespaceBundle, Map)} afterward should capture the snapshot once via
+ * {@link #getTopicFuturesInBundle(NamespaceBundle)} and pass the very same map to both calls: reusing one
+ * snapshot guarantees the cleanup step can neither miss a topic this call targeted, nor evict one it never
+ * observed (for example one installed by a newer ownership generation while this call was still running).
+ */
+ public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
+ boolean disconnectClients,
+ boolean closeWithoutWaitingClientDisconnect, long timeout, TimeUnit unit,
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures) {
CompletableFuture<Integer> future = unloadServiceUnit(
- serviceUnit, disconnectClients, closeWithoutWaitingClientDisconnect);
+ serviceUnit, disconnectClients, closeWithoutWaitingClientDisconnect, topicFutures);
ScheduledFuture<?> taskTimeout = executor().schedule(() -> {
if (!future.isDone()) {
log.warn().attr("serviceUnit", serviceUnit).log("Unloading of has timed out");
@@ -2907,50 +2923,50 @@ public class BrokerService implements Closeable {
*/
private CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
boolean disconnectClients,
- boolean closeWithoutWaitingClientDisconnect) {
+ boolean closeWithoutWaitingClientDisconnect,
+ Map<String, CompletableFuture<Optional<Topic>>>
+ topicFutures) {
List<CompletableFuture<Void>> closeFutures = new ArrayList<>();
- topics.forEach((name, topicFuture) -> {
+ topicFutures.forEach((name, topicFuture) -> {
TopicName topicName = TopicName.get(name);
- if (serviceUnit.includes(topicName)) {
- if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)
- && ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
- if (ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log)) {
- log.info()
- .attr("topic", topicName)
- .log("Skip unloading ExtensibleLoadManager internal topics. Such internal topic "
- + "should be closed when shutting down the broker.");
- }
- return;
+ if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)
+ && ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
+ if (ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log)) {
+ log.info()
+ .attr("topic", topicName)
+ .log("Skip unloading ExtensibleLoadManager internal topics. Such internal topic "
+ + "should be closed when shutting down the broker.");
}
+ return;
+ }
- // Topic needs to be unloaded
- log.info().attr("topic", topicName).log("Unloading topic");
- if (topicFuture.isCompletedExceptionally()) {
- try {
- topicFuture.get();
- } catch (InterruptedException | ExecutionException ex) {
- if (ex.getCause() instanceof ServiceUnitNotReadyException) {
- // Topic was already unloaded
- log.debug().attr("topic", topicName).log("Topic was already unloaded");
- return;
- } else {
- log.warn().attr("topic", topicName).exception(ex).log("Got exception when closing topic");
- }
+ // Topic needs to be unloaded
+ log.info().attr("topic", topicName).log("Unloading topic");
+ if (topicFuture.isCompletedExceptionally()) {
+ try {
+ topicFuture.get();
+ } catch (InterruptedException | ExecutionException ex) {
+ if (ex.getCause() instanceof ServiceUnitNotReadyException) {
+ // Topic was already unloaded
+ log.debug().attr("topic", topicName).log("Topic was already unloaded");
+ return;
+ } else {
+ log.warn().attr("topic", topicName).exception(ex).log("Got exception when closing topic");
}
}
- closeFutures.add(topicFuture
- .thenCompose(t -> t.isPresent() ? t.get().close(
- disconnectClients, closeWithoutWaitingClientDisconnect)
- : CompletableFuture.completedFuture(null))
- .exceptionally(e -> {
- if (e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException
- && e.getMessage().contains("Please redo the lookup")) {
- log.warn().attr("topic", topicName).log("Topic ownership check failed. Skipping it");
- return null;
- }
- throw FutureUtil.wrapToCompletionException(e);
- }));
}
+ closeFutures.add(topicFuture
+ .thenCompose(t -> t.isPresent() ? t.get().close(
+ disconnectClients, closeWithoutWaitingClientDisconnect)
+ : CompletableFuture.completedFuture(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException
+ && e.getMessage().contains("Please redo the lookup")) {
+ log.warn().attr("topic", topicName).log("Topic ownership check failed. Skipping it");
+ return null;
+ }
+ throw FutureUtil.wrapToCompletionException(e);
+ }));
});
if (getPulsar().getConfig().isTransactionCoordinatorEnabled()
@@ -2968,17 +2984,52 @@ public class BrokerService implements Closeable {
return FutureUtil.waitForAll(closeFutures).thenApply(v -> closeFutures.size());
}
+ /**
+ * Captures the topic futures currently cached for the given bundle. Call this before starting an unload so
+ * that a later {@link #cleanUnloadedTopicFromCache(NamespaceBundle, Map)} call can remove only the exact
+ * futures this unload observed, and never a newer ownership generation's entry installed afterward for the
+ * same bundle.
+ */
+ public Map<String, CompletableFuture<Optional<Topic>>> getTopicFuturesInBundle(NamespaceBundle serviceUnit) {
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures = new HashMap<>();
+ topics.forEach((name, topicFuture) -> {
+ if (serviceUnit.includes(TopicName.get(name))) {
+ topicFutures.put(name, topicFuture);
+ }
+ });
+ return topicFutures;
+ }
+
+ /**
+ * Cleans up topics that failed to unload from the broker's topic cache, using a snapshot captured at call
+ * time.
+ *
+ * @deprecated use {@link #cleanUnloadedTopicFromCache(NamespaceBundle, Map)} with the snapshot captured via
+ * {@link #getTopicFuturesInBundle(NamespaceBundle)} before the unload started, so that a stale cleanup can
+ * never evict a topic installed by a newer ownership generation while the unload was still running.
+ */
+ @Deprecated
public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit) {
- topics.forEach((topic, topicFuture) -> {
- TopicName topicName = TopicName.get(topic);
+ cleanUnloadedTopicFromCache(serviceUnit, getTopicFuturesInBundle(serviceUnit));
+ }
+
+ /**
+ * Cleans up topics that failed to unload from the broker's topic cache. Only removes a topic if its
+ * currently-cached future is exactly the one captured in {@code topicFutures} (see
+ * {@link #getTopicFuturesInBundle(NamespaceBundle)}), so a stale call for an old ownership generation can
+ * never evict a newer generation's topic.
+ */
+ public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit,
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures) {
+ topicFutures.forEach((topic, topicFuture) -> {
Optional<Topic> topicRef = extractTopic(topicFuture);
// The bundle cleanup should only remove topics already fenced by close/unload.
- if (serviceUnit.includes(topicName) && topicRef.isPresent() && isTopicBeingUnloaded(topicRef.get())) {
+ if (topicRef.isPresent() && isTopicBeingUnloaded(topicRef.get())) {
log.info()
.attr("value", serviceUnit.toString())
.attr("topic", topic)
.log("Clean unloaded topic from cache.");
- removeTopicFromCache(topicName.toString(), serviceUnit, topicFuture);
+ removeTopicFromCache(topic, serviceUnit, topicFuture);
}
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 760019c989dc..dd43938257ec 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -549,6 +549,35 @@ public class NamespaceServiceTest extends BrokerTestBase {
@SuppressWarnings("deprecation")
+ @Test
+ public void testSplitCompletesWhenReleasingTheOldBundleStalls() throws Exception {
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ OwnershipCache realOwnershipCache = namespaceService.getOwnershipCache();
+ OwnershipCache ownershipCache = spy(realOwnershipCache);
+ // The old bundle's release is queued behind an acquire that never settles (e.g. metadata store unreachable)
+ doReturn(new CompletableFuture<Void>()).when(ownershipCache).removeOwnership(any(NamespaceBundle.class));
+ namespaceService.setOwnershipCache(ownershipCache);
+
+ int originalTimeout = conf.getMetadataStoreOperationTimeoutSeconds();
+ conf.setMetadataStoreOperationTimeoutSeconds(2);
+ try {
+ NamespaceName nsname = NamespaceName.get("prop/ns-split-release-stalls");
+ admin.namespaces().createNamespace(nsname.toString());
+ TopicName topicName = TopicName.get("persistent://" + nsname + "/topic-1");
+ NamespaceBundles bundles = namespaceService.getNamespaceBundleFactory().getBundles(nsname);
+ NamespaceBundle bundle = bundles.findBundle(topicName);
+ ownershipCache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+
+ // The split itself succeeded; a release that never settles must not hold the split's completion
+ // hostage, it is bounded by the metadata operation timeout and then logged.
+ namespaceService.splitAndOwnBundle(bundle, false,
+ NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_ALGO, null).get(30, TimeUnit.SECONDS);
+ } finally {
+ conf.setMetadataStoreOperationTimeoutSeconds(originalTimeout);
+ namespaceService.setOwnershipCache(realOwnershipCache);
+ }
+ }
+
@Test
public void testSplitBundleAndRemoveOldBundleFromOwnerShipCache() throws Exception {
OwnershipCache ownershipCache = spy(pulsar.getNamespaceService().getOwnershipCache());
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 43e816aaa3a4..38a0f5bc0e22 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
@@ -21,28 +21,44 @@ package org.apache.pulsar.broker.namespace;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.doAnswer;
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.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Range;
import com.google.common.hash.Hashing;
+import java.util.ArrayList;
import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import lombok.Cleanup;
import org.apache.bookkeeper.common.util.OrderedScheduler;
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.Topic;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -50,6 +66,8 @@ import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
+import org.apache.pulsar.metadata.api.coordination.LockManager;
+import org.apache.pulsar.metadata.api.coordination.ResourceLock;
import org.apache.pulsar.metadata.api.extended.CreateOption;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
@@ -97,9 +115,14 @@ public class OwnershipCacheTest {
bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32());
nsService = mock(NamespaceService.class);
+ doReturn(CompletableFuture.completedFuture(null)).when(nsService)
+ .unloadNamespaceBundle(any(NamespaceBundle.class));
brokerService = mock(BrokerService.class);
doReturn(CompletableFuture.completedFuture(1)).when(brokerService)
.unloadServiceUnit(any(), anyBoolean(), anyBoolean(), anyLong(), any());
+ doReturn(CompletableFuture.completedFuture(1)).when(brokerService)
+ .unloadServiceUnit(any(), anyBoolean(), anyBoolean(), anyLong(), any(), any());
+ doReturn(Map.of()).when(brokerService).getTopicFuturesInBundle(any());
doReturn(config).when(pulsar).getConfiguration();
doReturn(nsService).when(pulsar).getNamespaceService();
@@ -107,6 +130,7 @@ public class OwnershipCacheTest {
config.setWebServicePort(Optional.empty());
doReturn(brokerService).when(pulsar).getBrokerService();
doReturn(selfBrokerUrl).when(pulsar).getBrokerServiceUrl();
+ doReturn(executor).when(pulsar).getExecutor();
}
@AfterMethod(alwaysRun = true)
@@ -407,4 +431,510 @@ public class OwnershipCacheTest {
assertNotNull(cache.getOwnedBundle(testFullBundle));
}
+ @Test
+ public void testStaleUnloadDoesNotReleaseReacquiredOwnership() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ doReturn(cache).when(nsService).getOwnershipCache();
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-stale-unload"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle staleOwnedBundle = cache.getOwnedBundle(bundle);
+ assertNotNull(staleOwnedBundle);
+
+ // Simulate the lock-expiry self-heal path: the expiry callback invalidated the local cache while the
+ // unload it triggered is still in flight, and a concurrent lookup re-acquires the bundle in between.
+ cache.invalidateLocalOwnerCache(bundle);
+ cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle reacquiredOwnedBundle = cache.getOwnedBundle(bundle);
+ assertNotNull(reacquiredOwnedBundle);
+ assertNotSame(reacquiredOwnedBundle, staleOwnedBundle);
+ ResourceLock<NamespaceEphemeralData> reacquiredLock = cache.getLocallyAcquiredLocks().get(bundle);
+ assertNotNull(reacquiredLock);
+
+ // The stale unload chain now runs its remaining steps against the old OwnedBundle instance.
+ staleOwnedBundle.handleUnloadRequest(pulsar, 5, TimeUnit.SECONDS).join();
+
+ // The re-acquired ownership must survive the stale unload untouched.
+ assertSame(cache.getLocallyAcquiredLocks().get(bundle), reacquiredLock);
+ assertTrue(store.exists(ServiceUnitUtils.path(bundle)).join());
+ assertTrue(reacquiredOwnedBundle.isActive());
+ assertTrue(cache.checkOwnershipAsync(bundle).get());
+ }
+
+ @Test
+ public void testTryAcquiringOwnershipWaitsForInFlightOwnedBundleRelease() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-ownedbundle-release-inflight"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle gen1 = cache.getOwnedBundle(bundle);
+ assertNotNull(gen1);
+
+ // Mirror what OwnedBundle.handleUnloadRequest does at the very end of a normal unload: release via the
+ // generation-aware, OwnedBundle-keyed overload. Do NOT wait for it to finish before racing an acquire
+ // against it below.
+ CompletableFuture<Void> removeFuture = cache.removeOwnership(gen1);
+
+ // A concurrent lookup racing the in-flight release must be queued behind it by the barrier, not see the
+ // stale, concurrently-releasing generation.
+ NamespaceEphemeralData reacquired = cache.tryAcquiringOwnership(bundle).get();
+ OwnedBundle afterReacquire = cache.getOwnedBundle(bundle);
+
+ removeFuture.get(10, TimeUnit.SECONDS);
+
+ assertNotSame(afterReacquire, gen1,
+ "tryAcquiringOwnership returned the stale, concurrently-releasing generation instead of a fresh one");
+ assertTrue(afterReacquire.isActive(), "reacquired OwnedBundle should be active");
+ assertSame(cache.getLocallyAcquiredLocks().get(bundle), afterReacquire.getResourceLock());
+ }
+
+ @Test
+ public void testRemoveOwnershipWithAcquisitionInFlight() throws Exception {
+ // Gate the lock acquisition so the test can invoke removeOwnership while the acquisition is in flight
+ LockManager<NamespaceEphemeralData> realLockManager =
+ coordinationService.getLockManager(NamespaceEphemeralData.class);
+ CompletableFuture<Void> gate = new CompletableFuture<>();
+ LockManager<NamespaceEphemeralData> gatedLockManager = new LockManager<NamespaceEphemeralData>() {
+ @Override
+ public CompletableFuture<Optional<NamespaceEphemeralData>> readLock(String path) {
+ return realLockManager.readLock(path);
+ }
+
+ @Override
+ public CompletableFuture<ResourceLock<NamespaceEphemeralData>> acquireLock(String path,
+ NamespaceEphemeralData value) {
+ return gate.thenCompose(__ -> realLockManager.acquireLock(path, value));
+ }
+
+ @Override
+ public CompletableFuture<List<String>> listLocks(String path) {
+ return realLockManager.listLocks(path);
+ }
+
+ @Override
+ public CompletableFuture<Void> asyncClose() {
+ return realLockManager.asyncClose();
+ }
+
+ @Override
+ public void close() throws Exception {
+ realLockManager.close();
+ }
+ };
+ CoordinationService gatedCoordinationService = mock(CoordinationService.class);
+ doReturn(gatedLockManager).when(gatedCoordinationService).getLockManager(NamespaceEphemeralData.class);
+ doReturn(gatedCoordinationService).when(pulsar).getCoordinationService();
+
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-inflight-remove"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ CompletableFuture<NamespaceEphemeralData> acquireFuture = cache.tryAcquiringOwnership(bundle);
+ assertFalse(acquireFuture.isDone());
+
+ CompletableFuture<Void> removeFuture = cache.removeOwnership(bundle);
+ gate.complete(null);
+ acquireFuture.join();
+ removeFuture.get(10, TimeUnit.SECONDS);
+
+ // After removeOwnership reported success and the in-flight acquisition settled, the broker must not
+ // silently retain (zombie) ownership.
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(cache.getLocallyAcquiredLocks().isEmpty());
+ assertTrue(cache.getOwnedBundles().isEmpty());
+ assertFalse(store.exists(ServiceUnitUtils.path(bundle)).join());
+ });
+ }
+
+ @Test
+ public void testExpiredLockIsRemovedFromLocallyAcquiredLocks() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-expired-lock"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get();
+ ResourceLock<NamespaceEphemeralData> lock = cache.getLocallyAcquiredLocks().get(bundle);
+ assertNotNull(lock);
+
+ // The lock dies without going through removeOwnership, like on a metadata session expiry
+ lock.release().join();
+
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(cache.getOwnedBundles().isEmpty());
+ assertTrue(cache.getLocallyAcquiredLocks().isEmpty());
+ });
+ verify(nsService).onNamespaceBundleUnload(bundle);
+ }
+
+ @Test
+ public void testExpiryBeforePublicationDoesNotLeaveActiveZombieOwnership() throws Exception {
+ // A ResourceLock whose expiry future is *already* completed by the time the loader attaches its
+ // lock-expiry listener: this deterministically forces the listener to run synchronously, inside the
+ // cache loader's thenApply, before the loader returns and the cache's own future for this bundle is
+ // published as done.
+ ResourceLock<NamespaceEphemeralData> alreadyExpiredLock = new ResourceLock<>() {
+ @Override
+ public String getPath() {
+ return "/dummy";
+ }
+
+ @Override
+ public NamespaceEphemeralData getValue() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<Void> updateValue(NamespaceEphemeralData newValue) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> release() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> getLockExpiredFuture() {
+ return CompletableFuture.completedFuture(null);
+ }
+ };
+ // Gate the lock acquisition itself so it settles only after tryAcquiringOwnership(bundle) has already
+ // returned to the caller: this ensures the cache has genuinely registered its (not yet done) future for
+ // the bundle before the loader's thenApply — and the already-expired lock's listener inside it — runs,
+ // matching how a real ZK acquisition callback fires on a different thread than the initial get() call.
+ CompletableFuture<Void> gate = new CompletableFuture<>();
+ LockManager<NamespaceEphemeralData> raceyLockManager = new LockManager<>() {
+ @Override
+ public CompletableFuture<Optional<NamespaceEphemeralData>> readLock(String path) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
+ @Override
+ public CompletableFuture<ResourceLock<NamespaceEphemeralData>> acquireLock(String path,
+ NamespaceEphemeralData value) {
+ return gate.thenApply(ignore -> alreadyExpiredLock);
+ }
+
+ @Override
+ public CompletableFuture<List<String>> listLocks(String path) {
+ return CompletableFuture.completedFuture(List.of());
+ }
+
+ @Override
+ public CompletableFuture<Void> asyncClose() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+ CoordinationService raceyCoordinationService = mock(CoordinationService.class);
+ doReturn(raceyLockManager).when(raceyCoordinationService).getLockManager(NamespaceEphemeralData.class);
+ doReturn(raceyCoordinationService).when(pulsar).getCoordinationService();
+
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-expiry-before-publication"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ CompletableFuture<NamespaceEphemeralData> acquireFuture = cache.tryAcquiringOwnership(bundle);
+ assertFalse(acquireFuture.isDone());
+
+ gate.complete(null);
+
+ // Expiry won the race: the acquisition itself must fail instead of publishing an OwnedBundle whose lock
+ // is already gone.
+ try {
+ acquireFuture.get(10, TimeUnit.SECONDS);
+ fail("acquisition should fail when the lock expired before ownership could be published");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof IllegalStateException);
+ }
+
+ // A failed load must not leave any trace behind: no cache entry claiming ownership, and no lock
+ // bookkeeping.
+ Awaitility.await().untilAsserted(() -> {
+ assertNull(cache.getOwnedBundle(bundle),
+ "cache still claims active ownership for a bundle whose lock expired before publication");
+ assertTrue(cache.getLocallyAcquiredLocks().isEmpty());
+ assertFalse(cache.checkOwnershipAsync(bundle).get());
+ });
+ // This generation was never published, so the ownership listeners never saw it owned; they must not be
+ // told it was unloaded either (an unmatched unload event can, for example, tear down the topic-policies
+ // cache of a namespace whose other bundles are still owned).
+ verify(nsService, never()).onNamespaceBundleOwned(bundle);
+ verify(nsService, never()).onNamespaceBundleUnload(bundle);
+ }
+
+ /**
+ * A {@link ResourceLock} whose expiry is driven by the test, so an old generation's lock-expiry listener can be
+ * made to run at an exact point: here, only after a newer generation has already been acquired.
+ */
+ private static class ControllableLock implements ResourceLock<NamespaceEphemeralData> {
+ private final CompletableFuture<Void> expiredFuture = new CompletableFuture<>();
+ private final AtomicBoolean released = new AtomicBoolean(false);
+
+ @Override
+ public String getPath() {
+ return "/controllable";
+ }
+
+ @Override
+ public NamespaceEphemeralData getValue() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<Void> updateValue(NamespaceEphemeralData newValue) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> release() {
+ released.set(true);
+ expiredFuture.complete(null);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> getLockExpiredFuture() {
+ return expiredFuture;
+ }
+ }
+
+ private static LockManager<NamespaceEphemeralData> lockManagerHandingOut(
+ List<ResourceLock<NamespaceEphemeralData>> locks) {
+ AtomicInteger next = new AtomicInteger();
+ return new LockManager<>() {
+ @Override
+ public CompletableFuture<Optional<NamespaceEphemeralData>> readLock(String path) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
+ @Override
+ public CompletableFuture<ResourceLock<NamespaceEphemeralData>> acquireLock(String path,
+ NamespaceEphemeralData value) {
+ return CompletableFuture.completedFuture(locks.get(next.getAndIncrement()));
+ }
+
+ @Override
+ public CompletableFuture<List<String>> listLocks(String path) {
+ return CompletableFuture.completedFuture(List.of());
+ }
+
+ @Override
+ public CompletableFuture<Void> asyncClose() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+ }
+
+ @Test
+ public void testDelayedExpiryListenerOfOldGenerationLeavesReacquiredOwnershipUntouched() throws Exception {
+ ControllableLock lock1 = new ControllableLock();
+ ControllableLock lock2 = new ControllableLock();
+ CoordinationService fakeCoordinationService = mock(CoordinationService.class);
+ doReturn(lockManagerHandingOut(List.of(lock1, lock2))).when(fakeCoordinationService)
+ .getLockManager(NamespaceEphemeralData.class);
+ doReturn(fakeCoordinationService).when(pulsar).getCoordinationService();
+
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ doReturn(cache).when(nsService).getOwnershipCache();
+ // Route the listener's unload through the real generation-blind lookup, exactly as NamespaceService does,
+ // so an unguarded listener really does tear down whatever generation is cached at that moment.
+ doAnswer(invocation -> {
+ NamespaceBundle b = invocation.getArgument(0);
+ OwnedBundle ob = cache.getOwnedBundle(b);
+ return ob == null
+ ? CompletableFuture.failedFuture(new IllegalStateException("not owned"))
+ : ob.handleUnloadRequest(pulsar, 5, TimeUnit.SECONDS);
+ }).when(nsService).unloadNamespaceBundle(any(NamespaceBundle.class));
+
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-delayed-expiry-listener"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+ OwnedBundle gen1 = cache.getOwnedBundle(bundle);
+ assertNotNull(gen1);
+ assertSame(gen1.getResourceLock(), lock1);
+
+ // Generation 1's lock died and its cache entry is gone, but its expiry listener has not run yet (it runs
+ // on whichever thread completes the expiry future). Meanwhile a lookup re-acquires the bundle.
+ cache.invalidateLocalOwnerCache(bundle);
+ cache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+ OwnedBundle gen2 = cache.getOwnedBundle(bundle);
+ assertNotNull(gen2);
+ assertNotSame(gen2, gen1);
+ assertSame(gen2.getResourceLock(), lock2);
+
+ // Now the delayed generation-1 listener finally runs.
+ lock1.expiredFuture.complete(null);
+ Awaitility.await().untilAsserted(() -> assertTrue(cache.getBundleOperationBarriers().isEmpty()));
+
+ // Generation 2 must survive: still cached, still active, its lock still registered and never released.
+ assertSame(cache.getOwnedBundle(bundle), gen2,
+ "old generation's expiry listener evicted the re-acquired generation from the cache");
+ assertTrue(gen2.isActive(), "old generation's expiry listener deactivated the re-acquired generation");
+ assertSame(cache.getLocallyAcquiredLocks().get(bundle), lock2);
+ assertFalse(lock2.released.get(), "old generation's expiry listener released the re-acquired lock");
+ assertTrue(cache.checkOwnershipAsync(bundle).get());
+ // ... and the ownership listeners must not be told that a bundle this broker still owns was unloaded
+ verify(nsService, never()).onNamespaceBundleUnload(bundle);
+ }
+
+ @Test
+ public void testSerializedOperationsDrainDeepBacklogWithoutWedgingBundle() throws Exception {
+ // Gate the first acquisition so a large backlog of operations queues up behind it on the barrier
+ LockManager<NamespaceEphemeralData> realLockManager =
+ coordinationService.getLockManager(NamespaceEphemeralData.class);
+ CompletableFuture<Void> gate = new CompletableFuture<>();
+ LockManager<NamespaceEphemeralData> gatedLockManager = new LockManager<>() {
+ @Override
+ public CompletableFuture<Optional<NamespaceEphemeralData>> readLock(String path) {
+ return realLockManager.readLock(path);
+ }
+
+ @Override
+ public CompletableFuture<ResourceLock<NamespaceEphemeralData>> acquireLock(String path,
+ NamespaceEphemeralData value) {
+ return gate.thenCompose(__ -> realLockManager.acquireLock(path, value));
+ }
+
+ @Override
+ public CompletableFuture<List<String>> listLocks(String path) {
+ return realLockManager.listLocks(path);
+ }
+
+ @Override
+ public CompletableFuture<Void> asyncClose() {
+ return realLockManager.asyncClose();
+ }
+
+ @Override
+ public void close() throws Exception {
+ realLockManager.close();
+ }
+ };
+ CoordinationService gatedCoordinationService = mock(CoordinationService.class);
+ doReturn(gatedLockManager).when(gatedCoordinationService).getLockManager(NamespaceEphemeralData.class);
+ doReturn(gatedCoordinationService).when(pulsar).getCoordinationService();
+
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-deep-backlog"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ CompletableFuture<NamespaceEphemeralData> first = cache.tryAcquiringOwnership(bundle);
+ assertFalse(first.isDone());
+
+ // Every one of these is a cache hit once the first acquire lands, i.e. completes synchronously when the
+ // barrier lets it run: the worst case for a drain that nests each operation inside the previous one.
+ int backlog = 5000;
+ List<CompletableFuture<NamespaceEphemeralData>> queued = new ArrayList<>(backlog);
+ for (int i = 0; i < backlog; i++) {
+ queued.add(cache.tryAcquiringOwnership(bundle));
+ }
+ gate.complete(null);
+ first.get(10, TimeUnit.SECONDS);
+
+ for (CompletableFuture<NamespaceEphemeralData> f : queued) {
+ assertNotNull(f.get(30, TimeUnit.SECONDS));
+ }
+ Awaitility.await().untilAsserted(() -> assertTrue(cache.getBundleOperationBarriers().isEmpty(),
+ "a barrier entry was left behind after all queued operations settled"));
+ // The bundle must still be usable afterward
+ cache.removeOwnership(bundle).get(10, TimeUnit.SECONDS);
+ Awaitility.await().untilAsserted(() -> assertTrue(cache.getBundleOperationBarriers().isEmpty()));
+ }
+
+ @Test
+ public void testUnloadRequestStillReleasesOwnershipWhenTopicUnloadThrowsSynchronously() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ doReturn(cache).when(nsService).getOwnershipCache();
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-unload-throws"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+ OwnedBundle ownedBundle = cache.getOwnedBundle(bundle);
+ assertNotNull(ownedBundle);
+
+ // e.g. the unload timeout could not be scheduled because the broker is shutting down. By then the topic
+ // closes have already been started from the snapshot, so the cleanup afterward must still use that
+ // snapshot rather than an empty one.
+ Map<String, CompletableFuture<Optional<Topic>>> snapshot =
+ Map.of("persistent://pulsar/ns-unload-throws/t", CompletableFuture.completedFuture(Optional.empty()));
+ doReturn(snapshot).when(brokerService).getTopicFuturesInBundle(bundle);
+ doThrow(new RejectedExecutionException("executor is shutting down")).when(brokerService)
+ .unloadServiceUnit(any(), anyBoolean(), anyBoolean(), anyLong(), any(), any());
+
+ // Must not throw synchronously: a synchronous failure after the isActive CAS would leave the bundle
+ // permanently inactive with its lock still registered, and a retried unload would fail fast forever.
+ CompletableFuture<Void> unloadFuture = ownedBundle.handleUnloadRequest(pulsar, 5, TimeUnit.SECONDS);
+ unloadFuture.get(10, TimeUnit.SECONDS);
+
+ // The topic-close failure is treated like any other unload failure: ownership is still released.
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(cache.getLocallyAcquiredLocks().isEmpty());
+ assertNull(cache.getOwnedBundle(bundle));
+ assertFalse(store.exists(ServiceUnitUtils.path(bundle)).join());
+ });
+ verify(brokerService).cleanUnloadedTopicFromCache(eq(bundle), same(snapshot));
+ }
+
+ @Test
+ public void testSerializedOperationsStillDrainWhenExecutorRejectsTasks() throws Exception {
+ // During shutdown the executor that hands the barrier to the next queued operation may already be gone;
+ // the queued release must still run rather than hang forever behind a barrier that is never released.
+ var rejectingExecutor = Executors.newSingleThreadScheduledExecutor();
+ rejectingExecutor.shutdownNow();
+ doReturn(rejectingExecutor).when(pulsar).getExecutor();
+
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-executor-rejects"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+ cache.removeOwnership(bundle).get(10, TimeUnit.SECONDS);
+
+ assertTrue(cache.getLocallyAcquiredLocks().isEmpty());
+ assertFalse(store.exists(ServiceUnitUtils.path(bundle)).join());
+ Awaitility.await().untilAsserted(() -> assertTrue(cache.getBundleOperationBarriers().isEmpty()));
+ }
+
+ @Test
+ public void testRemoveOwnershipOfDetachedOwnedBundleDoesNotReleaseCurrentOwnership() throws Exception {
+ OwnershipCache cache = new OwnershipCache(this.pulsar, nsService);
+ NamespaceBundle bundle = new NamespaceBundle(NamespaceName.get("pulsar/ns-detached-owned-bundle"),
+ Range.closedOpen(0L, (long) Integer.MAX_VALUE),
+ bundleFactory);
+
+ cache.tryAcquiringOwnership(bundle).get(10, TimeUnit.SECONDS);
+ ResourceLock<NamespaceEphemeralData> currentLock = cache.getLocallyAcquiredLocks().get(bundle);
+ assertNotNull(currentLock);
+
+ // An OwnedBundle that was not created by the cache is bound to no ownership generation, so releasing
+ // "its" ownership must not fall back to releasing whatever generation currently owns the bundle.
+ @SuppressWarnings("deprecation")
+ OwnedBundle detached = new OwnedBundle(bundle);
+ cache.removeOwnership(detached).get(10, TimeUnit.SECONDS);
+
+ assertSame(cache.getLocallyAcquiredLocks().get(bundle), currentLock);
+ assertTrue(store.exists(ServiceUnitUtils.path(bundle)).join());
+ assertTrue(cache.checkOwnershipAsync(bundle).get());
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 50cbbd5d4521..4a52b9b9bb10 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -1439,8 +1440,9 @@ public class BrokerServiceTest extends BrokerTestBase {
};
brokerService.addTopicEventListener(listener);
try {
+ Map<String, CompletableFuture<Optional<Topic>>> topicFutures = Map.of(topicName, oldTopicFuture);
CompletableFuture<Void> cleanupFuture = CompletableFuture.runAsync(
- () -> brokerService.cleanUnloadedTopicFromCache(bundle));
+ () -> brokerService.cleanUnloadedTopicFromCache(bundle, topicFutures));
assertTrue(oldTopicFuture.awaitIsDone(5, TimeUnit.SECONDS),
"cleanup should capture the old future before it is replaced");
@@ -1583,7 +1585,7 @@ public class BrokerServiceTest extends BrokerTestBase {
pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, false).get(5, TimeUnit.SECONDS);
((AbstractTopic) topic).isFenced = true;
try {
- brokerService.cleanUnloadedTopicFromCache(bundle);
+ brokerService.cleanUnloadedTopicFromCache(bundle, brokerService.getTopicFuturesInBundle(bundle));
assertFalse(brokerService.getTopics().containsKey(topicName),
"Inactive bundle cleanup should remove the matching topic future");
@@ -2316,6 +2318,87 @@ public class BrokerServiceTest extends BrokerTestBase {
assertFalse(MockTopicPoliciesService.FAILED_TOPICS.contains(topicName));
}
+ @Test
+ public void testCleanUnloadedTopicFromCacheIsGenerationSafe() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/staleCleanupTest-" + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ BrokerService brokerService = pulsar.getBrokerService();
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+
+ // Generation 1: load the topic and capture its future the way a real unload snapshot would.
+ Topic staleGenerationTopic = brokerService.getTopic(topicName, true).get().orElseThrow();
+ CompletableFuture<Optional<Topic>> staleGenerationFuture =
+ CompletableFuture.completedFuture(Optional.of(staleGenerationTopic));
+
+ // Simulate re-acquisition: the topic is closed and reloaded, installing a *new* generation's future/topic
+ // under the same name, while this broker still owns and serves the bundle.
+ admin.topics().unload(topicName);
+ Topic newGenerationTopic = brokerService.getTopic(topicName, true).get().orElseThrow();
+ assertNotSame(newGenerationTopic, staleGenerationTopic);
+
+ // A stale cleanup call for the old generation's unload arrives late. It must only ever act on the exact
+ // future it captured at unload start, never on a newer generation's entry for the same topic name.
+ brokerService.cleanUnloadedTopicFromCache(bundle, Map.of(topicName, staleGenerationFuture));
+
+ assertTrue(brokerService.getTopicReference(topicName).isPresent(),
+ "stale cleanup wrongly evicted the newer generation's topic from the cache");
+ }
+
+ @Test
+ public void testCleanUnloadedTopicFromCacheIsGenerationSafeForBookkeeping() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/staleCleanupBookkeepingTest-" + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ BrokerService brokerService = pulsar.getBrokerService();
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+
+ // Generation 1: load the topic and capture its future the way a real unload snapshot would.
+ Topic staleGenerationTopic = brokerService.getTopic(topicName, true).get().orElseThrow();
+ CompletableFuture<Optional<Topic>> staleGenerationFuture =
+ CompletableFuture.completedFuture(Optional.of(staleGenerationTopic));
+
+ // Simulate re-acquisition: the topic is closed and reloaded, installing a *new* generation's future/topic
+ // under the same name, while this broker still owns and serves the bundle.
+ admin.topics().unload(topicName);
+ brokerService.getTopic(topicName, true).get().orElseThrow();
+ // addTopicToStatsMaps() runs asynchronously off the topic-load future, so wait for it to land.
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(brokerService.getTopicStats(bundle).containsKey(topicName),
+ "the newer generation should be tracked in the per-bundle stats index after reload"));
+
+ // A stale cleanup call for the old generation's unload arrives late. The topics-map removal is a
+ // guarded no-op (proven by testCleanUnloadedTopicFromCacheIsGenerationSafe above), but the surrounding
+ // bookkeeping around it must be gated on that same guard too, not run unconditionally.
+ brokerService.cleanUnloadedTopicFromCache(bundle, Map.of(topicName, staleGenerationFuture));
+
+ assertTrue(brokerService.getTopicStats(bundle).containsKey(topicName),
+ "stale cleanup wrongly stripped the still-live newer generation's topic from the per-bundle "
+ + "stats index (multiLayerTopicsMap), even though the topics-map removal itself was "
+ + "correctly skipped");
+ }
+
+ @Test
+ public void testCleanUnloadedTopicFromCacheRemovesMatchingSnapshot() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/staleCleanupMatchTest-" + UUID.randomUUID();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ BrokerService brokerService = pulsar.getBrokerService();
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+ Topic topic = brokerService.getTopic(topicName, true).get().orElseThrow();
+ // The bundle cleanup only ever runs after unloadServiceUnit has attempted to close the topic, which
+ // fences it synchronously even on a failed/incomplete close. Mirror that precondition here.
+ ((AbstractTopic) topic).isFenced = true;
+
+ // A snapshot that matches exactly what is currently cached must still be cleaned up: this is the
+ // legitimate backstop case (a topic whose close() failed to remove itself from the cache).
+ Map<String, CompletableFuture<Optional<Topic>>> currentSnapshot = brokerService.getTopicFuturesInBundle(bundle);
+ brokerService.cleanUnloadedTopicFromCache(bundle, currentSnapshot);
+
+ assertFalse(brokerService.getTopicReference(topicName).isPresent(),
+ "cleanup should still remove a topic future that matches what was captured");
+ }
+
private static class BlockingCompletedFuture<T> extends CompletableFuture<T> {
private final CountDownLatch isDoneEntered = new CountDownLatch(1);
private final CountDownLatch allowIsDone = new CountDownLatch(1);