You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/07/03 12:06:07 UTC
[pulsar] 01/03: [improve][test] Add integration test for ExtensibleLoadManager (#20138)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e407c4b86a64e4eaa5ff291048d03c90cfb71035
Author: Kai Wang <kw...@apache.org>
AuthorDate: Fri Jun 16 09:11:59 2023 +0800
[improve][test] Add integration test for ExtensibleLoadManager (#20138)
---
.github/workflows/pulsar-ci.yaml | 4 +
build/run_integration_group.sh | 4 +
.../integration/containers/BrokerContainer.java | 4 +
.../integration/containers/PulsarContainer.java | 2 +-
.../loadbalance/ExtensibleLoadManagerTest.java | 390 +++++++++++++++++++++
.../src/test/resources/pulsar-loadbalance.xml | 28 ++
tests/integration/src/test/resources/pulsar.xml | 1 +
7 files changed, 432 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml
index 7139ea4cff9..35182f0bbd8 100644
--- a/.github/workflows/pulsar-ci.yaml
+++ b/.github/workflows/pulsar-ci.yaml
@@ -478,6 +478,10 @@ jobs:
- name: Messaging
group: MESSAGING
+ - name: LoadBalance
+ group: LOADBALANCE
+ no_coverage: true
+
- name: Shade on Java 8
group: SHADE_RUN
runtime_jdk: 8
diff --git a/build/run_integration_group.sh b/build/run_integration_group.sh
index bc1255d8d68..9372d1ecb93 100755
--- a/build/run_integration_group.sh
+++ b/build/run_integration_group.sh
@@ -156,6 +156,10 @@ test_group_messaging() {
mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-tls.xml -DintegrationTests
}
+test_group_loadbalance() {
+ mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-loadbalance.xml -DintegrationTests
+}
+
test_group_plugin() {
mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-plugin.xml -DintegrationTests
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java
index e0e85af024b..616d45554d7 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BrokerContainer.java
@@ -33,6 +33,10 @@ public class BrokerContainer extends PulsarContainer<BrokerContainer> {
tailContainerLog();
}
+ public String getHostName() {
+ return super.hostname;
+ }
+
@Override
protected void afterStart() {
DockerUtils.runCommandAsyncWithLogging(this.dockerClient, this.getContainerId(),
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
index dc11acd00c3..56d64ce5b2c 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
@@ -70,7 +70,7 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
public static final boolean PULSAR_CONTAINERS_LEAVE_RUNNING =
Boolean.parseBoolean(System.getenv("PULSAR_CONTAINERS_LEAVE_RUNNING"));
- private final String hostname;
+ protected final String hostname;
private final String serviceName;
private final String serviceEntryPoint;
private final int servicePort;
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java
new file mode 100644
index 00000000000..5fd2cba4153
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java
@@ -0,0 +1,390 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.loadbalance;
+
+import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_HTTP_PORT;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+import com.google.common.collect.Sets;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
+import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
+import org.apache.pulsar.common.policies.data.BundlesData;
+import org.apache.pulsar.common.policies.data.FailureDomain;
+import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.tests.TestRetrySupport;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+
+/**
+ * Integration tests for Pulsar ExtensibleLoadManagerImpl.
+ */
+@Slf4j
+public class ExtensibleLoadManagerTest extends TestRetrySupport {
+
+ private static final int NUM_BROKERS = 3;
+ private static final String DEFAULT_TENANT = "my-tenant";
+ private static final String DEFAULT_NAMESPACE = DEFAULT_TENANT + "/my-namespace";
+ private static final String nsSuffix = "-anti-affinity-enabled";
+
+ private final String clusterName = "MultiLoadManagerTest-" + UUID.randomUUID();
+ private final PulsarClusterSpec spec = PulsarClusterSpec.builder()
+ .clusterName(clusterName)
+ .numBrokers(NUM_BROKERS).build();
+ private PulsarCluster pulsarCluster = null;
+ private List<String> brokerUrls = null;
+ private String hosts;
+ private PulsarAdmin admin;
+
+ @BeforeClass(alwaysRun = true)
+ public void setup() throws Exception {
+ incrementSetupNumber();
+ Map<String, String> brokerEnvs = new HashMap<>();
+ brokerEnvs.put("loadManagerClassName",
+ "org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl");
+ brokerEnvs.put("loadBalancerLoadSheddingStrategy",
+ "org.apache.pulsar.broker.loadbalance.extensions.scheduler.TransferShedder");
+ brokerEnvs.put("forceDeleteNamespaceAllowed", "true");
+ brokerEnvs.put("loadBalancerDebugModeEnabled", "true");
+ brokerEnvs.put("PULSAR_MEM", "-Xmx512M");
+ spec.brokerEnvs(brokerEnvs);
+ pulsarCluster = PulsarCluster.forSpec(spec);
+ pulsarCluster.start();
+ brokerUrls = brokerUrls();
+
+ hosts = pulsarCluster.getAllBrokersHttpServiceUrl();
+ admin = PulsarAdmin.builder().serviceHttpUrl(hosts).build();
+ // all brokers alive
+ assertEquals(admin.brokers().getActiveBrokers(clusterName).size(), NUM_BROKERS);
+
+ admin.tenants().createTenant(DEFAULT_TENANT,
+ new TenantInfoImpl(new HashSet<>(), Set.of(pulsarCluster.getClusterName())));
+ admin.namespaces().createNamespace(DEFAULT_NAMESPACE, 100);
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void cleanup() {
+ markCurrentSetupNumberCleaned();
+ if (pulsarCluster != null) {
+ pulsarCluster.stop();
+ pulsarCluster = null;
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ public void startBroker() {
+ if (pulsarCluster != null) {
+ pulsarCluster.getBrokers().forEach(brokerContainer -> {
+ if (!brokerContainer.isRunning()) {
+ brokerContainer.start();
+ }
+ });
+ }
+ }
+
+ @Test(timeOut = 40 * 1000)
+ public void testConcurrentLookups() throws Exception {
+ String topicName = "persistent://" + DEFAULT_NAMESPACE + "/testConcurrentLookups";
+ List<PulsarAdmin> admins = new ArrayList<>();
+ int numAdminForBroker = 10;
+ for (String url : brokerUrls) {
+ for (int i = 0; i < numAdminForBroker; i++) {
+ admins.add(PulsarAdmin.builder().serviceHttpUrl(url).build());
+ }
+ }
+
+ admin.topics().createPartitionedTopic(topicName, 100);
+
+ var executor = Executors.newFixedThreadPool(admins.size());
+
+ CountDownLatch latch = new CountDownLatch(admins.size());
+ List<Map<String, String>> result = new CopyOnWriteArrayList<>();
+ for(var admin : admins) {
+ executor.execute(() -> {
+ try {
+ result.add(admin.lookups().lookupPartitionedTopic(topicName));
+ } catch (PulsarAdminException e) {
+ log.error("Lookup partitioned topic failed.", e);
+ }
+ latch.countDown();
+ });
+ }
+ latch.await();
+
+ assertEquals(result.size(), admins.size());
+
+ for (int i = 1; i < admins.size(); i++) {
+ assertEquals(result.get(i - 1), result.get(i));
+ }
+ }
+
+ @Test(timeOut = 30 * 1000)
+ public void testTransferAdminApi() throws Exception {
+ String topicName = "persistent://" + DEFAULT_NAMESPACE + "/testUnloadAdminApi";
+ admin.topics().createNonPartitionedTopic(topicName);
+ String broker = admin.lookups().lookupTopic(topicName);
+
+ int index = extractBrokerIndex(broker);
+
+ String bundleRange = admin.lookups().getBundleRange(topicName);
+
+ // Test transfer to current broker.
+ try {
+ admin.namespaces().unloadNamespaceBundle(DEFAULT_NAMESPACE, bundleRange, getBrokerUrl(index));
+ fail();
+ } catch (PulsarAdminException ex) {
+ assertTrue(ex.getMessage().contains("cannot be transfer to same broker"));
+ }
+
+ int transferToIndex = generateRandomExcludingX(NUM_BROKERS, index);
+ assertNotEquals(transferToIndex, index);
+ String transferTo = getBrokerUrl(transferToIndex);
+ admin.namespaces().unloadNamespaceBundle(DEFAULT_NAMESPACE, bundleRange, transferTo);
+
+ broker = admin.lookups().lookupTopic(topicName);
+
+ index = extractBrokerIndex(broker);
+ assertEquals(index, transferToIndex);
+ }
+
+ @Test(timeOut = 30 * 1000)
+ public void testSplitBundleAdminApi() throws Exception {
+ String topicName = "persistent://" + DEFAULT_NAMESPACE + "/testSplitBundleAdminApi";
+ admin.topics().createNonPartitionedTopic(topicName);
+ String broker = admin.lookups().lookupTopic(topicName);
+ log.info("The topic: {} owned by {}", topicName, broker);
+ BundlesData bundles = admin.namespaces().getBundles(DEFAULT_NAMESPACE);
+ int numBundles = bundles.getNumBundles();
+ var bundleRanges = bundles.getBoundaries().stream().map(Long::decode).sorted().toList();
+ String firstBundle = bundleRanges.get(0) + "_" + bundleRanges.get(1);
+ admin.namespaces().splitNamespaceBundle(DEFAULT_NAMESPACE, firstBundle, true, null);
+ BundlesData bundlesData = admin.namespaces().getBundles(DEFAULT_NAMESPACE);
+
+ long mid = bundleRanges.get(0) + (bundleRanges.get(1) - bundleRanges.get(0)) / 2;
+
+ assertEquals(bundlesData.getNumBundles(), numBundles + 1);
+ String lowBundle = String.format("0x%08x", bundleRanges.get(0));
+ String midBundle = String.format("0x%08x", mid);
+ String highBundle = String.format("0x%08x", bundleRanges.get(1));
+ assertTrue(bundlesData.getBoundaries().contains(lowBundle));
+ assertTrue(bundlesData.getBoundaries().contains(midBundle));
+ assertTrue(bundlesData.getBoundaries().contains(highBundle));
+
+ // Test split bundle with invalid bundle range.
+ try {
+ admin.namespaces().splitNamespaceBundle(DEFAULT_NAMESPACE, "invalid", true, null);
+ fail();
+ } catch (PulsarAdminException ex) {
+ assertTrue(ex.getMessage().contains("Invalid bundle range"));
+ }
+ }
+
+ @Test(timeOut = 30 * 1000)
+ public void testDeleteNamespace() throws Exception {
+ String namespace = DEFAULT_TENANT + "/test-delete-namespace";
+ String topicName = "persistent://" + namespace + "/test-delete-namespace-topic";
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet(clusterName));
+ assertTrue(admin.namespaces().getNamespaces(DEFAULT_TENANT).contains(namespace));
+ admin.topics().createPartitionedTopic(topicName, 2);
+ String broker = admin.lookups().lookupTopic(topicName);
+ log.info("The topic: {} owned by: {}", topicName, broker);
+ admin.namespaces().deleteNamespace(namespace, true);
+ assertFalse(admin.namespaces().getNamespaces(DEFAULT_TENANT).contains(namespace));
+ }
+
+ @Test(timeOut = 40 * 1000)
+ public void testStopBroker() throws Exception {
+ String topicName = "persistent://" + DEFAULT_NAMESPACE + "/test-stop-broker-topic";
+
+ admin.topics().createNonPartitionedTopic(topicName);
+ String broker = admin.lookups().lookupTopic(topicName);
+ log.info("The topic: {} owned by: {}", topicName, broker);
+
+ int idx = extractBrokerIndex(broker);
+ for (BrokerContainer container : pulsarCluster.getBrokers()) {
+ String name = container.getHostName();
+ if (name.contains(String.valueOf(idx))) {
+ container.stop();
+ }
+ }
+
+ String broker1 = admin.lookups().lookupTopic(topicName);
+
+ assertNotEquals(broker1, broker);
+ }
+
+ @Test(timeOut = 40 * 1000)
+ public void testAntiaffinityPolicy() throws PulsarAdminException {
+ final String namespaceAntiAffinityGroup = "my-anti-affinity-filter";
+ final String antiAffinityEnabledNameSpace = DEFAULT_TENANT + "/my-ns-filter" + nsSuffix;
+ final int numPartition = 20;
+
+ List<String> activeBrokers = admin.brokers().getActiveBrokers();
+
+ assertEquals(activeBrokers.size(), NUM_BROKERS);
+
+ for (int i = 0; i < activeBrokers.size(); i++) {
+ String namespace = antiAffinityEnabledNameSpace + "-" + i;
+ admin.namespaces().createNamespace(namespace, 10);
+ admin.namespaces().setNamespaceAntiAffinityGroup(namespace, namespaceAntiAffinityGroup);
+ admin.clusters().createFailureDomain(clusterName, namespaceAntiAffinityGroup, FailureDomain.builder()
+ .brokers(Set.of(activeBrokers.get(i))).build());
+ }
+
+ Set<String> result = new HashSet<>();
+ for (int i = 0; i < activeBrokers.size(); i++) {
+ final String topic = "persistent://" + antiAffinityEnabledNameSpace + "-" + i +"/topic";
+ admin.topics().createPartitionedTopic(topic, numPartition);
+
+ Map<String, String> topicToBroker = admin.lookups().lookupPartitionedTopic(topic);
+
+ assertEquals(topicToBroker.size(), numPartition);
+
+ HashSet<String> brokers = new HashSet<>(topicToBroker.values());
+
+ assertEquals(brokers.size(), 1);
+ result.add(brokers.iterator().next());
+ log.info("Topic: {}, lookup result: {}", topic, brokers.iterator().next());
+ }
+
+ assertEquals(result.size(), NUM_BROKERS);
+ }
+
+ @Test(timeOut = 40 * 1000)
+ public void testIsolationPolicy() throws PulsarAdminException {
+ final String namespaceIsolationPolicyName = "my-isolation-policy";
+ final String isolationEnabledNameSpace = DEFAULT_TENANT + "/my-isolation-policy" + nsSuffix;
+ Map<String, String> parameters1 = new HashMap<>();
+ parameters1.put("min_limit", "1");
+ parameters1.put("usage_threshold", "100");
+
+ List<String> activeBrokers = admin.brokers().getActiveBrokers();
+
+ assertEquals(activeBrokers.size(), NUM_BROKERS);
+
+ admin.namespaces().createNamespace(isolationEnabledNameSpace);
+ admin.clusters().createNamespaceIsolationPolicy(clusterName, namespaceIsolationPolicyName, NamespaceIsolationData
+ .builder()
+ .namespaces(List.of(isolationEnabledNameSpace))
+ .autoFailoverPolicy(AutoFailoverPolicyData.builder()
+ .policyType(AutoFailoverPolicyType.min_available)
+ .parameters(parameters1)
+ .build())
+ .primary(List.of(getHostName(0)))
+ .secondary(List.of(getHostName(1)))
+ .build());
+ final String topic = "persistent://" + isolationEnabledNameSpace + "/topic";
+ admin.topics().createNonPartitionedTopic(topic);
+
+ String broker = admin.lookups().lookupTopic(topic);
+
+ for (BrokerContainer container : pulsarCluster.getBrokers()) {
+ String name = container.getHostName();
+ if (name.contains("0")) {
+ container.stop();
+ }
+ }
+
+ assertEquals(extractBrokerIndex(broker), 0);
+
+ broker = admin.lookups().lookupTopic(topic);
+
+ assertEquals(extractBrokerIndex(broker), 1);
+
+ for (BrokerContainer container : pulsarCluster.getBrokers()) {
+ String name = container.getHostName();
+ if (name.contains("1")) {
+ container.stop();
+ }
+ }
+ try {
+ admin.lookups().lookupTopic(topic);
+ fail();
+ } catch (Exception ex) {
+ log.error("Failed to lookup topic: ", ex);
+ assertTrue(ex.getMessage().contains("Failed to look up a broker"));
+ }
+ }
+
+ private String getBrokerUrl(int index) {
+ return String.format("pulsar-broker-%d:%d", index, BROKER_HTTP_PORT);
+ }
+
+ private String getHostName(int index) {
+ return String.format("pulsar-broker-%d", index);
+ }
+
+ private int extractBrokerIndex(String broker) {
+ String pattern = "pulsar://.*-(\\d+):\\d+";
+ Pattern compiledPattern = Pattern.compile(pattern);
+ Matcher matcher = compiledPattern.matcher(broker);
+ if (!matcher.find()){
+ throw new IllegalArgumentException("Failed to extract broker index");
+ }
+ return Integer.parseInt(matcher.group(1));
+ }
+
+ private int generateRandomExcludingX(int n, int x) {
+ Random random = new Random();
+ int randomNumber;
+
+ do {
+ randomNumber = random.nextInt(n);
+ } while (randomNumber == x);
+
+ return randomNumber;
+ }
+
+ private List<String> brokerUrls() {
+ Collection<BrokerContainer> brokers = pulsarCluster.getBrokers();
+ List<String> brokerUrls = new ArrayList<>(NUM_BROKERS);
+ brokers.forEach(broker -> {
+ brokerUrls.add("http://" + broker.getHost() + ":" + broker.getMappedPort(BROKER_HTTP_PORT));
+ });
+ return brokerUrls;
+ }
+}
diff --git a/tests/integration/src/test/resources/pulsar-loadbalance.xml b/tests/integration/src/test/resources/pulsar-loadbalance.xml
new file mode 100644
index 00000000000..dfc4536e255
--- /dev/null
+++ b/tests/integration/src/test/resources/pulsar-loadbalance.xml
@@ -0,0 +1,28 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<!DOCTYPE suite SYSTEM "https://testng.org/testng-1.0.dtd" >
+<suite name="Pulsar (LoadBalance) Integration Tests" verbose="2" annotations="JDK">
+ <test name="loadbalance-test-suite" preserve-order="true">
+ <classes>
+ <class name="org.apache.pulsar.tests.integration.loadbalance.ExtensibleLoadManagerTest"/>
+ </classes>
+ </test>
+</suite>
\ No newline at end of file
diff --git a/tests/integration/src/test/resources/pulsar.xml b/tests/integration/src/test/resources/pulsar.xml
index 7fbe9fa24d8..d309a7695dd 100644
--- a/tests/integration/src/test/resources/pulsar.xml
+++ b/tests/integration/src/test/resources/pulsar.xml
@@ -30,6 +30,7 @@
<suite-file path="./tiered-filesystem-storage.xml"/>
<suite-file path="./pulsar-function.xml" />
<suite-file path="./pulsar-messaging.xml" />
+ <suite-file path="./pulsar-loadbalance.xml" />
<suite-file path="./pulsar-backwards-compatibility.xml" />
<suite-file path="./pulsar-io-sinks.xml" />
<suite-file path="./pulsar-io-sources.xml" />