Skip to content

[fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup #26197

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SongOf wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix][broker] Fix ownership-generation races in OwnershipCache removeOwnership and lock-expiry cleanup #26197

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 20 additions & 3 deletions .../org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1071,20 +1072,36 @@ private CompletableFuture<Integer> 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<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;
})
.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
Expand All @@ -1094,7 +1111,7 @@ private CompletableFuture<Integer> 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)
Expand Down
20 changes: 16 additions & 4 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand Down
59 changes: 49 additions & 10 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnedBundle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -27,15 +29,26 @@
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
@CustomLog
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.
Expand All @@ -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.
Expand All @@ -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<NamespaceEphemeralData> resourceLock) {
this.bundle = suName;
this.resourceLock = resourceLock;
IS_ACTIVE_UPDATER.set(this, TRUE);
}

ResourceLock<NamespaceEphemeralData> getResourceLock() {
return resourceLock;
}

/**
* Access to the namespace name.
*
Expand Down Expand Up @@ -130,11 +153,27 @@ public CompletableFuture<Void> 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<String, CompletableFuture<Optional<Topic>>> 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
Expand All @@ -146,12 +185,12 @@ public CompletableFuture<Void> 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;
})
Comment on lines 173 to 190

@Denovo1998 Denovo1998 Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The lock release is now generation-safe, but the stale unload is still bundle-wide at the topic layer.

BrokerService.unloadServiceUnit operates on the broker's current topic map, and cleanUnloadedTopicFromCache later scans the current map again and eventually calls topics.remove(topic) because no create future is supplied. If this unload belongs to an expired OwnedBundle and the bundle is re-acquired while the old close futures are still running, the stale cleanup can remove a topic future installed by the newer ownership generation.

The new test mocks BrokerService, so it only verifies that the new lock/znode/OwnedBundle survives and cannot detect topic-cache corruption.

Could we either prevent re-acquisition until the previous generation's unload cleanup finishes, or make the cleanup identity-safe by capturing the topic futures at unload start and removing them with topics.remove(topicName, capturedFuture)?

@SongOf SongOf Jul 22, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Denovo1998
Fixed — thanks for the detailed repro path, that's exactly what was happening.
We went with the identity-safe cleanup option, but implemented it as a single shared snapshot rather than two independent scans (capturing once for unload, again for cleanup would leave its own gap — see below):

  • BrokerService.getTopicFuturesInBundle(NamespaceBundle) captures (topicName → topicFuture) for the bundle in one pass, at unload start.
  • unloadServiceUnit now has an overload that consumes that snapshot directly instead of re-scanning topics, so the set of topics it closes and the set cleanUnloadedTopicFromCache later validates against are guaranteed to be identical — no second scan, no window for a topic to be missed by one side and caught by the other.
  • cleanUnloadedTopicFromCache now takes that same snapshot and removes via topics.remove(topic, capturedFuture) instead of the old null-future call, so a stale cleanup can only ever touch the exact future it observed at unload start. A newer generation's topic future is a different object and simply won't match the CAS.
  • Both call sites — OwnedBundle.handleUnloadRequest and ServiceUnitStateChannelImpl.closeServiceUnit — capture the snapshot once and thread it through both calls.

On re-acquisition: we didn't block it. Blocking re-acquisition until the previous generation's cleanup finishes would add a synchronization point across OwnershipCache and BrokerService and slow down bundle flapping recovery; identity-safe removal gets full correctness without that coupling.

On the test: agreed the mocked-BrokerService test can't see this class of bug. Added BrokerServiceTest#testCleanUnloadedTopicFromCacheIsGenerationSafe against a real embedded broker — it loads a topic, closes+reloads it to install a new generation's future under the same name (standing in for the re-acquire), then invokes the stale cleanup with the old snapshot and asserts the new generation's topic is still present. Confirmed it fails without the identity check (topics.remove(topic, null)-equivalent) and passes with it.

While implementing this we also caught a follow-on issue in review: unloadServiceUnit's topic-closing scan and the pre-captured cleanup snapshot were originally two separate scans of topics, which reopened a smaller version of the same gap for any topic created in between. Fixed by making unloadServiceUnit consume the caller-supplied snapshot directly (see above) instead of doing its own scan — now there's exactly one point-in-time view shared by both steps.

.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));
Expand Down
Loading