Skip to content

[fix][broker] Fix ownership-generation races in OwnershipCache remove… #1

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 2 commits into
base: master
Choose a base branch
from
Open

[fix][broker] Fix ownership-generation races in OwnershipCache remove… #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
724a40a
[fix][broker] Fix ownership-generation races in OwnershipCache remove…
Jul 16, 2026
c1211bc
Merge branch 'master' into fix/broker-ownership-cache-generation-race
Jul 27, 2026
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 @@ -1085,20 +1086,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 @@ -1108,7 +1125,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 @@ -1070,12 +1070,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 @@ -1087,7 +1099,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;
})
.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
Loading