You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/17 02:15:37 UTC
(pulsar) branch master updated: [fix][broker] Handle flow_or_qps bundle split requests in ELM (#25442)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 c78a99e6b46c [fix][broker] Handle flow_or_qps bundle split requests in ELM (#25442)
c78a99e6b46c is described below
commit c78a99e6b46c553330c1bf1d774e760793effe1d
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Oct 17 10:15:31 2026 +0800
[fix][broker] Handle flow_or_qps bundle split requests in ELM (#25442)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
Co-authored-by: Matteo Merli <mm...@apache.org>
---
.../extensions/ExtensibleLoadManagerImpl.java | 68 +++++++++++++---------
.../pulsar/broker/namespace/NamespaceService.java | 11 ++--
.../apache/pulsar/broker/admin/AdminApiTest.java | 18 ++++++
.../extensions/ExtensibleLoadManagerImplTest.java | 16 +++++
4 files changed, 82 insertions(+), 31 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 0e78fd682f0a..d49b1a6347f6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -759,34 +759,50 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager, BrokerS
final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle.toString());
NamespaceBundle namespaceBundle =
pulsar.getNamespaceService().getNamespaceBundleFactory().getBundle(namespaceName, bundleRange);
- return pulsar.getNamespaceService().getSplitBoundary(namespaceBundle, splitAlgorithm, boundaries)
- .thenCompose(splitBundlesPair -> {
- if (splitBundlesPair == null) {
- String msg = format("Bundle %s not found under namespace", namespaceBundle);
- log.error(msg);
- return FutureUtil.failedFuture(new IllegalStateException(msg));
+ return pulsar.getNamespaceService().getSplitBoundary(namespaceBundle, boundaries, splitAlgorithm)
+ .thenCompose(splitBoundaries -> {
+ if (splitBoundaries == null || splitBoundaries.isEmpty()) {
+ log.info()
+ .attr("namespaceObject", namespaceBundle.getNamespaceObject().toString())
+ .attr("boundaries", boundaries)
+ .attr("bundle", namespaceBundle.getBundleRange())
+ .log("No valid boundary found to split bundle");
+ return CompletableFuture.completedFuture(null);
}
-
- return getOwnershipAsync(Optional.empty(), bundle)
- .thenCompose(brokerOpt -> {
- if (brokerOpt.isEmpty()) {
- String msg = String.format("Namespace bundle: %s is not owned by any broker.",
- bundle);
- log.warn(msg);
- throw new IllegalStateException(msg);
+ return pulsar.getNamespaceService().getNamespaceBundleFactory()
+ .splitBundles(namespaceBundle, splitBoundaries.size() + 1, splitBoundaries)
+ .thenCompose(splitBundlesPair -> {
+ if (splitBundlesPair == null) {
+ String msg = format("Bundle %s not found under namespace", namespaceBundle);
+ log.error(msg);
+ return FutureUtil.failedFuture(new IllegalStateException(msg));
}
- String sourceBroker = brokerOpt.get();
- SplitDecision splitDecision = new SplitDecision();
- List<NamespaceBundle> splitBundles = splitBundlesPair.getRight();
- Map<String, Optional<String>> splitServiceUnitToDestBroker = new HashMap<>();
- splitBundles.forEach(splitBundle -> splitServiceUnitToDestBroker
- .put(splitBundle.getBundleRange(), Optional.empty()));
- splitDecision.setSplit(
- new Split(bundle.toString(), sourceBroker, splitServiceUnitToDestBroker));
- splitDecision.setLabel(Success);
- splitDecision.setReason(Admin);
- return splitAsync(splitDecision,
- conf.getNamespaceBundleUnloadingTimeoutMs(), TimeUnit.MILLISECONDS);
+
+ return getOwnershipAsync(Optional.empty(), bundle)
+ .thenCompose(brokerOpt -> {
+ if (brokerOpt.isEmpty()) {
+ String msg = String.format(
+ "Namespace bundle: %s is not owned by any broker.",
+ bundle);
+ log.warn(msg);
+ throw new IllegalStateException(msg);
+ }
+ String sourceBroker = brokerOpt.get();
+ SplitDecision splitDecision = new SplitDecision();
+ List<NamespaceBundle> splitBundles = splitBundlesPair.getRight();
+ Map<String, Optional<String>> splitServiceUnitToDestBroker =
+ new HashMap<>();
+ splitBundles.forEach(splitBundle -> splitServiceUnitToDestBroker
+ .put(splitBundle.getBundleRange(), Optional.empty()));
+ splitDecision.setSplit(
+ new Split(bundle.toString(), sourceBroker,
+ splitServiceUnitToDestBroker));
+ splitDecision.setLabel(Success);
+ splitDecision.setReason(Admin);
+ return splitAsync(splitDecision,
+ conf.getNamespaceBundleUnloadingTimeoutMs(),
+ TimeUnit.MILLISECONDS);
+ });
});
});
}
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 4d39c226de22..a3351a8621b3 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
@@ -88,6 +88,7 @@ import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.lookup.data.LookupData;
import org.apache.pulsar.common.naming.BundleSplitOption;
+import org.apache.pulsar.common.naming.FlowOrQpsEquallyDivideBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.FlowOrQpsEquallyDivideBundleSplitOption;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
@@ -981,7 +982,7 @@ public class NamespaceService implements AutoCloseable {
CompletableFuture<Void> completionFuture,
NamespaceBundleSplitAlgorithm splitAlgorithm,
List<Long> boundaries) {
- BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);
+ BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config, splitAlgorithm);
splitAlgorithm.getSplitBoundary(bundleSplitOption).whenComplete((splitBoundaries, ex) -> {
CompletableFuture<List<NamespaceBundle>> updateFuture = new CompletableFuture<>();
@@ -1180,16 +1181,16 @@ public class NamespaceService implements AutoCloseable {
public CompletableFuture<List<Long>> getSplitBoundary(
NamespaceBundle bundle, List<Long> boundaries, NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm) {
- BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);
+ BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config, nsBundleSplitAlgorithm);
return nsBundleSplitAlgorithm.getSplitBoundary(bundleSplitOption);
}
private BundleSplitOption getBundleSplitOption(NamespaceBundle bundle,
List<Long> boundaries,
- ServiceConfiguration config) {
+ ServiceConfiguration config,
+ NamespaceBundleSplitAlgorithm splitAlgorithm) {
BundleSplitOption bundleSplitOption;
- if (config.getDefaultNamespaceBundleSplitAlgorithm()
- .equals(NamespaceBundleSplitAlgorithm.FLOW_OR_QPS_EQUALLY_DIVIDE)) {
+ if (splitAlgorithm instanceof FlowOrQpsEquallyDivideBundleSplitAlgorithm) {
Map<String, TopicStatsImpl> topicStatsMap = pulsar.getBrokerService().getTopicStats(bundle);
bundleSplitOption = new FlowOrQpsEquallyDivideBundleSplitOption(this, bundle, boundaries,
topicStatsMap,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 88f36d2bacec..beadfc0649ef 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -1749,6 +1749,24 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
}
}
+ @Test
+ public void testNamespaceSplitBundleWithFlowOrQpsAlgorithmUsesRequestedAlgorithm() throws Exception {
+ final String namespace = "prop-xyz/flow-or-qps-split";
+ final String topicName = "persistent://" + namespace + "/topic-1";
+ admin.namespaces().createNamespace(namespace, 1);
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.lookups().lookupTopic(topicName);
+
+ try {
+ admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0xffffffff", true,
+ NamespaceBundleSplitAlgorithm.FLOW_OR_QPS_EQUALLY_DIVIDE);
+ } catch (Exception e) {
+ fail("split bundle with flow_or_qps_equally_divide shouldn't have thrown exception", e);
+ }
+
+ assertEquals(admin.namespaces().getBundles(namespace).getNumBundles(), 1);
+ }
+
@Test
public void testNamespacesGetTopicHashPositions() throws Exception {
// Force to create a namespace with only one bundle and create a topic
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index c93bc8ec75b8..fbfb3ac5b9ca 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -123,6 +123,7 @@ import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.common.naming.NamespaceBundle;
+import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.naming.TopicDomain;
@@ -1034,6 +1035,21 @@ public class ExtensibleLoadManagerImplTest extends ExtensibleLoadManagerImplBase
assertTrue(bundlesData.getBoundaries().contains(midBundle));
assertTrue(bundlesData.getBoundaries().contains(highBundle));
}
+
+ @Test(timeOut = 30 * 1000)
+ public void testSplitBundleWithFlowOrQpsAdminAPINoValidBoundary() throws Exception {
+ String namespace = "public/test-split-with-flow-or-qps";
+ String topic = "persistent://" + namespace + "/test-split-with-flow-or-qps";
+ admin.namespaces().createNamespace(namespace, 1);
+ admin.topics().createNonPartitionedTopic(topic);
+ admin.lookups().lookupTopic(topic);
+
+ admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0xffffffff", true,
+ NamespaceBundleSplitAlgorithm.FLOW_OR_QPS_EQUALLY_DIVIDE);
+
+ assertEquals(admin.namespaces().getBundles(namespace).getNumBundles(), 1);
+ }
+
@Test(timeOut = 30 * 1000)
public void testDeleteNamespaceBundle() throws Exception {
final String namespace = "public/testDeleteNamespaceBundle";