You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2023/05/11 14:37:43 UTC
[pulsar] 02/02: [improve] [broker] Skip split boundle if only one broker (#20190)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f3edfc578fab3d74298615429efeb2aaa72a4d32
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sat Apr 29 01:54:05 2023 +0800
[improve] [broker] Skip split boundle if only one broker (#20190)
Co-authored-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit d135c4a115038dc61f8fe2d230cb1f0c02239f92)
---
.../loadbalance/impl/ModularLoadManagerImpl.java | 2 +-
.../pulsar/broker/stats/PrometheusMetricsTest.java | 56 +++++++++++++++++++++
.../pulsar/client/api/BrokerServiceLookupTest.java | 58 ++++++++++++++++++++++
3 files changed, 115 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index d81f6949f43..293ff2760f0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -700,7 +700,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
public void checkNamespaceBundleSplit() {
if (!conf.isLoadBalancerAutoBundleSplitEnabled() || pulsar.getLeaderElectionService() == null
- || !pulsar.getLeaderElectionService().isLeader()) {
+ || !pulsar.getLeaderElectionService().isLeader() || loadData.getBrokerData().size() <= 1) {
return;
}
final boolean unloadSplitBundles = pulsar.getConfiguration().isLoadBalancerAutoUnloadSplitBundlesEnabled();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index 473adc26daa..103b8f94df1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -36,6 +36,7 @@ import java.nio.charset.StandardCharsets;
import java.text.NumberFormat;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@@ -74,6 +75,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.compaction.Compactor;
+import org.apache.zookeeper.CreateMode;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -548,6 +550,58 @@ public class PrometheusMetricsTest extends BrokerTestBase {
c1.acknowledge(c1.receive());
}
+ // Mock another broker to make split task work.
+ String json =
+ "{"
+ + "\"webServiceUrl\": \"http://127.0.0.1:0\","
+ + "\"webServiceUrlTls\": \"https://127.0.0.1:0\","
+ + "\"pulsarServiceUrl\": \"pulsar://127.0.0.1:0\","
+ + "\"pulsarServiceUrlTls\": \"pulsar+ssl://127.0.0.1:0\","
+ + "\"persistentTopicsEnabled\": true,"
+ + "\"nonPersistentTopicsEnabled\": true,"
+ + "\"cpu\": {"
+ + "\"usage\": 0.0,"
+ + "\"limit\": 1000.0"
+ + "},"
+ + "\"memory\": {"
+ + "\"usage\": 124.1398696899414,"
+ + "\"limit\": 1024.0"
+ + "},"
+ + "\"directMemory\": {"
+ + "\"usage\": 4.0,"
+ + "\"limit\": 1024.0"
+ + "},"
+ + "\"bandwidthIn\": {"
+ + "\"usage\": -1.0,"
+ + "\"limit\": -1.0"
+ + "},"
+ + "\"bandwidthOut\": {"
+ + "\"usage\": -1.0,"
+ + "\"limit\": -1.0"
+ + "},"
+ + "\"msgThroughputIn\": 0.0,"
+ + "\"msgThroughputOut\": 0.0,"
+ + "\"msgRateIn\": 0.0,"
+ + "\"msgRateOut\": 0.0,"
+ + "\"lastUpdate\": 1683812593521,"
+ + "\"lastStats\": {},"
+ + "\"numTopics\": 0,"
+ + "\"numBundles\": 0,"
+ + "\"numConsumers\": 0,"
+ + "\"numProducers\": 0,"
+ + "\"bundles\": [],"
+ + "\"lastBundleGains\": [],"
+ + "\"lastBundleLosses\": [],"
+ + "\"brokerVersionString\": \"2.10.4\","
+ + "\"protocols\": {},"
+ + "\"advertisedListeners\": {},"
+ + "\"maxResourceUsage\": 0.00390625,"
+ + "\"loadReportType\": \"LocalBrokerData\","
+ + "\"bundleStats\": {}"
+ + "}";
+ String mockedBroker = "/loadbalance/brokers/127.0.0.1:0";
+ mockZooKeeper.create(mockedBroker, json.getBytes(), Collections.emptyList(), CreateMode.EPHEMERAL);
+
pulsar.getBrokerService().updateRates();
Awaitility.await().untilAsserted(() -> assertTrue(pulsar.getBrokerService().getBundleStats().size() > 0));
ModularLoadManagerWrapper loadManager = (ModularLoadManagerWrapper)pulsar.getLoadManager().get();
@@ -572,6 +626,8 @@ public class PrometheusMetricsTest extends BrokerTestBase {
assertTrue(metrics.containsKey("pulsar_lb_bandwidth_out_usage"));
assertTrue(metrics.containsKey("pulsar_lb_bundles_split_count"));
+
+ mockZooKeeper.delete(mockedBroker, 0);
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
index e3bb6a92dc0..396fda82b59 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
@@ -66,6 +66,7 @@ import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import lombok.Cleanup;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
@@ -78,6 +79,7 @@ import org.apache.pulsar.broker.loadbalance.impl.SimpleResourceUnit;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.common.naming.NamespaceBundle;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
@@ -762,6 +764,62 @@ public class BrokerServiceLookupTest extends ProducerConsumerBase {
}
}
+ @Test(timeOut = 20000)
+ public void testSkipSplitBundleIfOnlyOneBroker() throws Exception {
+
+ log.info("-- Starting {} test --", methodName);
+ final String loadBalancerName = conf.getLoadManagerClassName();
+ final int defaultNumberOfNamespaceBundles = conf.getDefaultNumberOfNamespaceBundles();
+ final int loadBalancerNamespaceBundleMaxTopics = conf.getLoadBalancerNamespaceBundleMaxTopics();
+
+ final String namespace = "my-property/my-ns";
+ final String topicName1 = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/tp_");
+ final String topicName2 = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/tp_");
+ try {
+ // configure broker with ModularLoadManager.
+ stopBroker();
+ conf.setDefaultNumberOfNamespaceBundles(1);
+ conf.setLoadBalancerNamespaceBundleMaxTopics(1);
+ conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
+ startBroker();
+ final ModularLoadManagerWrapper modularLoadManagerWrapper =
+ (ModularLoadManagerWrapper) pulsar.getLoadManager().get();
+ final ModularLoadManagerImpl modularLoadManager =
+ (ModularLoadManagerImpl) modularLoadManagerWrapper.getLoadManager();
+
+ // Create one topic and trigger tasks, then verify there is only one bundle now.
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName1)
+ .subscriptionName("my-subscriber-name").subscribe();
+ List<NamespaceBundle> bounldes1 = pulsar.getNamespaceService().getNamespaceBundleFactory()
+ .getBundles(NamespaceName.get(namespace)).getBundles();
+ pulsar.getBrokerService().updateRates();
+ pulsar.getLoadManager().get().writeLoadReportOnZookeeper();
+ pulsar.getLoadManager().get().writeResourceQuotasToZooKeeper();
+ modularLoadManager.updateAll();
+ assertEquals(bounldes1.size(), 1);
+
+ // Create the second topic and trigger tasks, then verify the split task will be skipped.
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName2)
+ .subscriptionName("my-subscriber-name").subscribe();
+ pulsar.getBrokerService().updateRates();
+ pulsar.getLoadManager().get().writeLoadReportOnZookeeper();
+ pulsar.getLoadManager().get().writeResourceQuotasToZooKeeper();
+ modularLoadManager.updateAll();
+ List<NamespaceBundle> bounldes2 = pulsar.getNamespaceService().getNamespaceBundleFactory()
+ .getBundles(NamespaceName.get(namespace)).getBundles();
+ assertEquals(bounldes2.size(), 1);
+
+ consumer1.close();
+ consumer2.close();
+ admin.topics().delete(topicName1, false);
+ admin.topics().delete(topicName2, false);
+ } finally {
+ conf.setDefaultNumberOfNamespaceBundles(defaultNumberOfNamespaceBundles);
+ conf.setLoadBalancerNamespaceBundleMaxTopics(loadBalancerNamespaceBundleMaxTopics);
+ conf.setLoadManagerClassName(loadBalancerName);
+ }
+ }
+
@Test(timeOut = 10000)
public void testPartitionedMetadataWithDeprecatedVersion() throws Exception {