Skip to content

[fix][broker] Avoid split non-existent bundle #25031

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
coderzc wants to merge 3 commits into
base: master
Choose a base branch
from
+59 −0
Open

[fix][broker] Avoid split non-existent bundle #25031

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c08ff26
[fix][load] Avoid get old loadData during split bundle
coderzc Dec 2, 2025
d582d6b
Only skip split the bundles that do not exist in Metadata
coderzc Dec 9, 2025
eeaaf73
Update bundle existence check to use NamespaceBundles instead of meta…
coderzc Dec 9, 2025
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
13 changes: 13 additions & 0 deletions ...roker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.policies.data.ResourceQuota;
Expand Down Expand Up @@ -368,6 +369,11 @@ public CompletableFuture<Set<String>> getAvailableBrokersAsync() {
return future;
}

public boolean checkBundleDataExistInNamespaceBundles(NamespaceBundles namespaceBundles, String bundleRange) {
List<NamespaceBundle> bundles = namespaceBundles.getBundles();
return bundles.stream().anyMatch(b -> b.getBundleRange().equals(bundleRange));
}

// Attempt to local the data for the given bundle in metadata store
// If it cannot be found, return the default bundle data.
@Override
Expand Down Expand Up @@ -767,6 +773,13 @@ public void checkNamespaceBundleSplit() {
continue;
}

NamespaceBundles bundles = namespaceBundleFactory.getBundles(NamespaceName.get(namespaceName));
if (!checkBundleDataExistInNamespaceBundles(bundles, bundleRange)) {
log.warn("Bundle {} has been removed, skip split this bundle ",
bundleName);
continue;
}

// Make sure the same bundle is not selected again.
loadData.getBundleData().remove(bundleName);
localData.getLastStats().remove(bundleName);
Expand Down
46 changes: 46 additions & 0 deletions ...r/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1130,4 +1130,50 @@ public void testRemoveNonExistBundleData()
assertFalse(bundlesAfterSplit.contains(bundleWillBeSplit.getBundleRange()));
}

@Test
public void testRepeatSplitBundle() throws Exception {
final String cluster = "use";
final String tenant = "my-tenant";
final String namespace = "repeat-split-bundle";
final String topicName = tenant + "/" + namespace + "/" + "topic";
int bundleNumbers = 8;

admin1.clusters().createCluster(cluster, ClusterData.builder()
.serviceUrl(pulsar1.getWebServiceAddress()).build());
admin1.tenants().createTenant(tenant,
new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet(cluster)));
admin1.namespaces().createNamespace(tenant + "/" + namespace, bundleNumbers);

LoadData loadData = (LoadData) getField(primaryLoadManager, "loadData");
LocalBrokerData localData = (LocalBrokerData) getField(primaryLoadManager, "localData");

@Cleanup
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsar1.getBrokerServiceUrl()).build();

// create a lot of topic to fully distributed among bundles.
for (int i = 0; i < 10; i++) {
String topicNameI = topicName + i;
admin1.topics().createPartitionedTopic(topicNameI, 20);
// trigger bundle assignment

pulsarClient.newConsumer().topic(topicNameI)
.subscriptionName("my-subscriber-name2").subscribe();
}

String topicToFindBundle = topicName + 0;
NamespaceBundle realBundle = pulsar1.getNamespaceService().getBundle(TopicName.get(topicToFindBundle));
String bundleKey = realBundle.toString();
log.info("Before bundle={}", bundleKey);

NamespaceBundleStats stats = new NamespaceBundleStats();
stats.msgRateIn = 100000.0;
localData.getLastStats().put(bundleKey, stats);
pulsar1.getBrokerService().updateRates();

primaryLoadManager.updateAll();

primaryLoadManager.updateAll();
Assert.assertFalse(loadData.getBundleData().containsKey(bundleKey));
}

}
Loading