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 3f5bd2569faf9..5834fa4c61d78 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.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; @@ -1071,12 +1072,28 @@ private CompletableFuture closeServiceUnit(String serviceUnit, boolean 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>> 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; @@ -1084,7 +1101,7 @@ private CompletableFuture closeServiceUnit(String serviceUnit, boolean .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 @@ -1094,7 +1111,7 @@ private CompletableFuture closeServiceUnit(String serviceUnit, boolean .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 91632c1ee4cd7..1c0313f065494 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 @@ -1052,12 +1052,24 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle, // 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. + return pulsar.getNamespaceService().getOwnershipCache().removeOwnership(bundle) + .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 @@ -1069,7 +1081,7 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle, .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; 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 71fd8efa1aec9..c1580108459f6 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.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 @@ 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 resourceLock; + /** * {@link #nsLock} is used to protect read/write access to {@link #isActive} flag and the corresponding code section * based on {@link #isActive} flag. @@ -55,9 +68,8 @@ public class OwnedBundle { * @param suName */ public OwnedBundle(NamespaceBundle suName) { - this.bundle = suName; - IS_ACTIVE_UPDATER.set(this, TRUE); - }; + this(suName, true); + } /** * Constructor to allow set initial active flag. @@ -67,9 +79,20 @@ public OwnedBundle(NamespaceBundle suName) { */ public OwnedBundle(NamespaceBundle suName, boolean active) { this.bundle = suName; + this.resourceLock = null; IS_ACTIVE_UPDATER.set(this, active ? TRUE : FALSE); } + OwnedBundle(NamespaceBundle suName, ResourceLock resourceLock) { + this.bundle = suName; + this.resourceLock = resourceLock; + IS_ACTIVE_UPDATER.set(this, TRUE); + } + + ResourceLock getResourceLock() { + return resourceLock; + } + /** * Access to the namespace name. * @@ -130,11 +153,27 @@ public CompletableFuture handleUnloadRequest(PulsarService pulsar, long ti 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. + Map>> topicFutures = + pulsar.getBrokerService().getTopicFuturesInBundle(bundle); + // 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. + return pulsar.getBrokerService().unloadServiceUnit( + bundle, true, closeWithoutWaitingClientDisconnect, timeout, timeoutUnit, topicFutures) .handle((numUnloadedTopics, ex) -> { if (ex != null) { // ignore topic-close failure to unload bundle @@ -146,12 +185,12 @@ public CompletableFuture handleUnloadRequest(PulsarService pulsar, long ti 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 bc37d1f990620..5bfe975d1a887 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 @@ -33,6 +33,9 @@ import java.util.concurrent.Executor; 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.common.naming.NamespaceBundle; @@ -73,6 +76,22 @@ public class OwnershipCache { private final Map> 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> bundleOperationBarriers = new ConcurrentHashMap<>(); + /** * The loading cache of locally owned NamespaceBundle objects. */ @@ -92,14 +111,53 @@ public CompletableFuture asyncLoad(NamespaceBundle namespaceBundle, return lockManager.acquireLock(ServiceUnitUtils.path(namespaceBundle), selfOwnerInfo) .thenApply(rl -> { locallyAcquiredLocks.put(namespaceBundle, rl); + OwnedBundle ownedBundle = new OwnedBundle(namespaceBundle, rl); + // Set by the expiry listener below once it runs, whether that happens synchronously as + // part of registering it on the next line (the lock was already expired at that point) + // or later, asynchronously on whatever thread completes the future. thenRun() only + // guarantees inline, synchronous execution for the former case: per the CompletableFuture + // class javadoc, a non-async dependent action "may be performed by the thread that + // completes the current CompletableFuture, or by any other caller of a completion + // method" — there is no guarantee that a *concurrent* completion is reflected here by + // the time thenRun() returns. So this flag alone cannot be trusted to catch every case + // where the lock died right around registration; the isDone() recheck below closes that + // gap by reading the future's own state directly instead of relying on the listener + // having finished running. + AtomicBoolean expiredBeforePublication = new AtomicBoolean(false); rl.getLockExpiredFuture() .thenRun(() -> { log.info().attr("path", rl.getPath()).log("Resource lock has expired"); - namespaceService.unloadNamespaceBundle(namespaceBundle); - invalidateLocalOwnerCache(namespaceBundle); + expiredBeforePublication.set(true); + locallyAcquiredLocks.remove(namespaceBundle, rl); + 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; + }); + invalidateLocalOwnerCache(namespaceBundle, ownedBundle); namespaceService.onNamespaceBundleUnload(namespaceBundle); }); - return new OwnedBundle(namespaceBundle); + if (expiredBeforePublication.get() || 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. The isDone() check is a defensive addition alongside the flag: it reads + // the future's state directly, so it also catches the case where the lock expired + // concurrently with registration above but the listener callback (and therefore the + // flag) hasn't finished running yet. Even without it, a lock that expires around here + // and is missed by both checks still converges correctly once the listener does run + // (see invalidateLocalOwnerCache(NamespaceBundle, OwnedBundle)) — this just narrows + // that window rather than being the sole safeguard against it. + throw new IllegalStateException( + "Lock for bundle " + namespaceBundle + + " expired before ownership could be published"); + } + return ownedBundle; }); } } @@ -209,13 +267,14 @@ public CompletableFuture tryAcquiringOwnership(Namespace 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 +282,86 @@ public CompletableFuture tryAcquiringOwnership(Namespace * */ public CompletableFuture removeOwnership(NamespaceBundle bundle) { - ResourceLock 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 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. + * + *

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 CompletableFuture serialize(NamespaceBundle bundle, Supplier> operation) { + CompletableFuture result = new CompletableFuture<>(); + CompletableFuture opDone = new CompletableFuture<>(); + AtomicReference> 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); + } + 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. + * + *

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. + * + *

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 removeOwnership(OwnedBundle ownedBundle) { + ResourceLock lock = ownedBundle.getResourceLock(); if (lock == null) { - // We don't own the specified bundle anymore - return CompletableFuture.completedFuture(null); + // The instance is not bound to a lock (not created by this cache): fall back to removing whatever + // ownership currently exists for the bundle, which is itself serialized already. + return removeOwnership(ownedBundle.getNamespaceBundle()); } - - 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,6 +475,33 @@ public void invalidateLocalOwnerCache(NamespaceBundle namespaceBundle) { 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. + * + *

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) { + CompletableFuture future = ownedBundlesCache.getIfPresent(namespaceBundle); + if (future == null) { + return; + } + future.whenComplete((ownedBundle, ex) -> { + if (ex == null && ownedBundle == expectedOwnedBundle) { + ownedBundlesCache.asMap().remove(namespaceBundle, future); + } + }); + } + @VisibleForTesting public Map> getLocallyAcquiredLocks() { return locallyAcquiredLocks; 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 1ccd1478daf9e..f3f6e2e3910d7 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 @@ -2619,8 +2619,24 @@ public CompletableFuture checkTopicNsOwnership(final String topic) { public CompletableFuture 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 unloadServiceUnit(NamespaceBundle serviceUnit, + boolean disconnectClients, + boolean closeWithoutWaitingClientDisconnect, long timeout, TimeUnit unit, + Map>> topicFutures) { CompletableFuture 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"); @@ -2644,50 +2660,50 @@ public CompletableFuture unloadServiceUnit(NamespaceBundle serviceUnit, */ private CompletableFuture unloadServiceUnit(NamespaceBundle serviceUnit, boolean disconnectClients, - boolean closeWithoutWaitingClientDisconnect) { + boolean closeWithoutWaitingClientDisconnect, + Map>> + topicFutures) { List> 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() @@ -2705,17 +2721,39 @@ private CompletableFuture unloadServiceUnit(NamespaceBundle serviceUnit return FutureUtil.waitForAll(closeFutures).thenApply(v -> closeFutures.size()); } - public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit) { - for (String topic : topics.keySet()) { - TopicName topicName = TopicName.get(topic); - if (serviceUnit.includes(topicName) && getTopicReference(topic).isPresent()) { + /** + * 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>> getTopicFuturesInBundle(NamespaceBundle serviceUnit) { + Map>> 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. 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>> topicFutures) { + topicFutures.forEach((topic, topicFuture) -> { + if (getTopicReference(topic).isPresent()) { log.info() .attr("value", serviceUnit.toString()) .attr("topic", topic) .log("Clean unloaded topic from cache."); - pulsar.getBrokerService().removeTopicFromCache(topicName.toString(), serviceUnit, null); + removeTopicFromCache(topic, serviceUnit, topicFuture); } - } + }); } public AuthorizationService getAuthorizationService() { @@ -2739,6 +2777,18 @@ public CompletableFuture removeTopicFromCache(AbstractTopic topic) { private void removeTopicFromCache(String topic, NamespaceBundle namespaceBundle, CompletableFuture> createTopicFuture) { + // Gate everything below on the identity-guarded removal itself: if the currently-cached future no + // longer matches createTopicFuture (a newer ownership generation already replaced it), none of the + // bookkeeping or events below may run either, or a stale cleanup call would still strip a still-live + // newer generation's multiLayerTopicsMap/replication-metrics/compactor-stats/segment-load bookkeeping + // and fire spurious UNLOAD events for a topic that was never actually removed. + boolean removed = createTopicFuture == null + ? topics.remove(topic) != null + : topics.remove(topic, createTopicFuture); + if (!removed) { + return; + } + String bundleName = namespaceBundle.toString(); String namespaceName = TopicName.get(topic).getNamespaceObject().toString(); @@ -2767,12 +2817,6 @@ private void removeTopicFromCache(String topic, NamespaceBundle namespaceBundle, } } - if (createTopicFuture == null) { - topics.remove(topic); - } else { - topics.remove(topic, createTopicFuture); - } - Compactor compactor = pulsar.getNullableCompactor(); if (compactor != null) { compactor.getStats().removeTopic(topic); 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 43e816aaa3a4b..5c6e98b071f01 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 @@ -27,12 +27,16 @@ 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.EnumSet; +import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -50,6 +54,8 @@ 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 +103,14 @@ public void setup() throws Exception { 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(); @@ -407,4 +418,240 @@ public void testReestablishOwnership() throws Exception { 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 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 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 realLockManager = + coordinationService.getLockManager(NamespaceEphemeralData.class); + CompletableFuture gate = new CompletableFuture<>(); + LockManager gatedLockManager = new LockManager() { + @Override + public CompletableFuture> readLock(String path) { + return realLockManager.readLock(path); + } + + @Override + public CompletableFuture> acquireLock(String path, + NamespaceEphemeralData value) { + return gate.thenCompose(__ -> realLockManager.acquireLock(path, value)); + } + + @Override + public CompletableFuture> listLocks(String path) { + return realLockManager.listLocks(path); + } + + @Override + public CompletableFuture 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 acquireFuture = cache.tryAcquiringOwnership(bundle); + assertFalse(acquireFuture.isDone()); + + CompletableFuture 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 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()); + }); + } + + @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 alreadyExpiredLock = new ResourceLock<>() { + @Override + public String getPath() { + return "/dummy"; + } + + @Override + public NamespaceEphemeralData getValue() { + return null; + } + + @Override + public CompletableFuture updateValue(NamespaceEphemeralData newValue) { + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture release() { + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture 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 gate = new CompletableFuture<>(); + LockManager raceyLockManager = new LockManager<>() { + @Override + public CompletableFuture> readLock(String path) { + return CompletableFuture.completedFuture(Optional.empty()); + } + + @Override + public CompletableFuture> acquireLock(String path, + NamespaceEphemeralData value) { + return gate.thenApply(ignore -> alreadyExpiredLock); + } + + @Override + public CompletableFuture> listLocks(String path) { + return CompletableFuture.completedFuture(List.of()); + } + + @Override + public CompletableFuture 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 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()); + }); + } + } 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 4a6e329c72f81..65797b5d800a3 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 @@ -31,6 +31,7 @@ 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; @@ -2051,6 +2052,84 @@ public void testGetTopicWhenTopicPoliciesFail() throws Exception { 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> 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> 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)); + brokerService.getTopic(topicName, true).get(); + + // 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>> 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"); + } + static class MockTopicPoliciesService extends TopicPoliciesService.TopicPoliciesServiceDisabled { static final Set FAILED_TOPICS = ConcurrentHashMap.newKeySet();