Skip to content

[ci][improve][test] Improve AdminApiTest to reduce the execution time #10

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

Closed
coderzc wants to merge 2 commits into from
Closed

[ci][improve][test] Improve AdminApiTest to reduce the execution time #10

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
feb5e21
improve AdminApiTest
coderzc Oct 10, 2022
7d502b1
Merge branch 'apache:master' into slow/AdminApiTest
coderzc Oct 11, 2022
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
131 changes: 100 additions & 31 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,9 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

Expand All @@ -163,7 +164,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {

private NamespaceBundleFactory bundleFactory;

@BeforeMethod
@BeforeClass
@Override
public void setup() throws Exception {
conf.setSystemTopicEnabled(false);
Expand Down Expand Up @@ -192,14 +193,39 @@ public void setup() throws Exception {
otherPulsar = mockPulsarSetup.getPulsar();
otheradmin = mockPulsarSetup.getAdmin();

// Setup namespaces
setupClusters();
}

@AfterMethod(alwaysRun = true)
public void resetClusters() throws Exception {
pulsar.getConfiguration().setForceDeleteTenantAllowed(true);
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
for (String tenant : admin.tenants().getTenants()) {
for (String namespace : admin.namespaces().getNamespaces(tenant)) {
deleteNamespaceGraceFully(namespace, true);
}
admin.tenants().deleteTenant(tenant, true);
}

for (String cluster : admin.clusters().getClusters()) {
admin.clusters().deleteCluster(cluster);
}

pulsar.getConfiguration().setForceDeleteTenantAllowed(false);
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);

resetConfig();
setupClusters();
}

private void setupClusters() throws PulsarAdminException {
admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
admin.tenants().createTenant("prop-xyz", tenantInfo);
admin.namespaces().createNamespace("prop-xyz/ns1", Set.of("test"));
}

@AfterMethod(alwaysRun = true)
@AfterClass(alwaysRun = true)
@Override
public void cleanup() throws Exception {
adminTls.close();
Expand Down Expand Up @@ -1190,6 +1216,7 @@ public void testGetStats() throws Exception {
String subName = "my-sub";

// create consumer and subscription
@Cleanup
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName(subName).subscribe();
TopicStats topicStats = admin.topics().getStats(topic, false, false, true);

Expand Down Expand Up @@ -1227,7 +1254,9 @@ public void testGetPartitionedStatsInternal() throws Exception {
assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 2);

// create consumer and subscription
pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe();
@Cleanup
Consumer<byte[]> consumer =
pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName(subName).subscribe();

// publish several messages
publishMessagesOnPersistentTopic(partitionedTopicName, 10);
Expand Down Expand Up @@ -1426,8 +1455,10 @@ public void testForceDeleteTenantNotAllowed() throws Exception {

@Test
public void testNamespaceSplitBundle() throws Exception {
admin.namespaces().createNamespace("prop-xyz/splitBundle", Set.of("test"));

// Force to create a topic
final String namespace = "prop-xyz/ns1";
final String namespace = "prop-xyz/splitBundle";
final String topicName = (new StringBuilder("persistent://")).append(namespace).append("/ds2").toString();
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topicName)
Expand Down Expand Up @@ -1489,7 +1520,9 @@ public void testNamespaceSplitBundleWithTopicCountEquallyDivideAlgorithm() throw
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
}
producers.forEach(Producer::closeAsync);
for (Producer<byte[]> producer : producers) {
producer.close();
}
}

@Test
Expand Down Expand Up @@ -1684,6 +1717,9 @@ public void testNamespaceSplitBundleWithInvalidAlgorithm() {

@Test
public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm() throws Exception {
cleanup();
setup();

conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.TOPIC_COUNT_EQUALLY_DIVIDE);
// Force to create a topic
final String namespace = "prop-xyz/ns1";
Expand Down Expand Up @@ -1717,7 +1753,9 @@ public void testNamespaceSplitBundleWithDefaultTopicCountEquallyDivideAlgorithm(
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertNotEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
}
producers.forEach(Producer::closeAsync);
for (Producer<byte[]> producer : producers) {
producer.close();
}
conf.setDefaultNamespaceBundleSplitAlgorithm(NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_NAME);
}

Expand Down Expand Up @@ -1819,22 +1857,24 @@ public void testNamespaceSplitBundleConcurrent() throws Exception {

@Test
public void testNamespaceUnloadBundle() throws Exception {
assertEquals(admin.topics().getList("prop-xyz/ns1"), new ArrayList<>());
admin.namespaces().createNamespace("prop-xyz/unloadBundle", Set.of("test"));

assertEquals(admin.topics().getList("prop-xyz/unloadBundle"), new ArrayList<>());

// Force to create a topic
publishMessagesOnPersistentTopic("persistent://prop-xyz/ns1/ds2", 0);
assertEquals(admin.topics().getList("prop-xyz/ns1"),
Lists.newArrayList("persistent://prop-xyz/ns1/ds2"));
publishMessagesOnPersistentTopic("persistent://prop-xyz/unloadBundle/ds2", 0);
assertEquals(admin.topics().getList("prop-xyz/unloadBundle"),
Lists.newArrayList("persistent://prop-xyz/unloadBundle/ds2"));

// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1/ds2")
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/unloadBundle/ds2")
.subscriptionName("my-sub").subscribe();
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/ns1/ds2"),
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/unloadBundle/ds2"),
Lists.newArrayList("my-sub"));

// Create producer
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/ns1/ds2")
.topic("persistent://prop-xyz/unloadBundle/ds2")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
Expand All @@ -1847,13 +1887,13 @@ public void testNamespaceUnloadBundle() throws Exception {
producer.close();

try {
admin.namespaces().unloadNamespaceBundle("prop-xyz/ns1", "0x00000000_0xffffffff");
admin.namespaces().unloadNamespaceBundle("prop-xyz/unloadBundle", "0x00000000_0xffffffff");
} catch (Exception e) {
fail("Unload shouldn't have throw exception");
}

// check that no one owns the namespace
NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/ns1"),
NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/unloadBundle"),
Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED));
assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle));
assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle));
Expand All @@ -1863,14 +1903,17 @@ public void testNamespaceUnloadBundle() throws Exception {

// Force reload of namespace and wait for topic to be ready
Awaitility.await().timeout(30, TimeUnit.SECONDS).ignoreExceptionsInstanceOf(PulsarAdminException.class)
.until(() -> admin.topics().getStats("persistent://prop-xyz/ns1/ds2") != null);
.until(() -> admin.topics().getStats("persistent://prop-xyz/unloadBundle/ds2") != null);

admin.topics().deleteSubscription("persistent://prop-xyz/ns1/ds2", "my-sub");
admin.topics().delete("persistent://prop-xyz/ns1/ds2");
admin.topics().deleteSubscription("persistent://prop-xyz/unloadBundle/ds2", "my-sub");
admin.topics().delete("persistent://prop-xyz/unloadBundle/ds2");
}

@Test(dataProvider = "numBundles")
public void testNamespaceBundleUnload(Integer numBundles) throws Exception {
cleanup();
setup();

admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles);
admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Set.of("test"));

Expand Down Expand Up @@ -1976,15 +2019,25 @@ public void testClearBacklogOnNamespace(Integer numBundles) throws Exception {
admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns1-bundles", Set.of("test"));

// create consumer and subscription
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub")
.subscribe();
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-1")
.subscribe();
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-2")
@Cleanup
Consumer<byte[]> consumer =
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub")
.subscribe();
@Cleanup
Consumer<byte[]> consumer2 =
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-1")
.subscribe();
@Cleanup
Consumer<byte[]> consumer3 =
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2").subscriptionName("my-sub-2")
.subscribe();
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub")
@Cleanup
Consumer<byte[]> consumer4 =
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub")
.subscribe();
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1")
@Cleanup
Consumer<byte[]> consumer5 =
pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1").subscriptionName("my-sub-1")
.subscribe();

// Create producer
Expand Down Expand Up @@ -2048,7 +2101,8 @@ public void testUnsubscribeOnNamespace(Integer numBundles) throws Exception {
.subscriptionName("my-sub").subscribe();
Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2")
.subscriptionName("my-sub-1").subscribe();
/* Consumer consumer3 = */ pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2")
@Cleanup
Consumer<byte[]> consumer3 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds2")
.subscriptionName("my-sub-2").subscribe();
Consumer<byte[]> consumer4 = pulsarClient.newConsumer().topic("persistent://prop-xyz/ns1-bundles/ds1")
.subscriptionName("my-sub").subscribe();
Expand Down Expand Up @@ -2833,7 +2887,9 @@ public void testPulsarAdminForUriAndUrlEncoding(String topicName) throws Excepti
final int numOfPartitions = 4;
admin.topics().createPartitionedTopic(topic1, numOfPartitions);
// Create a consumer to get stats on this topic
pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe();
@Cleanup
Consumer<byte[]> consumer =
pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe();

TopicsImpl persistent = (TopicsImpl) admin.topics();
Field field = TopicsImpl.class.getDeclaredField("adminV2Topics");
Expand Down Expand Up @@ -3089,12 +3145,14 @@ public void testCompactionStatus() throws Exception {
public void testTopicStatsLastExpireTimestampForSubscription() throws PulsarAdminException, PulsarClientException, InterruptedException {
admin.namespaces().setNamespaceMessageTTL("prop-xyz/ns1", 10);
final String topic = "persistent://prop-xyz/ns1/testTopicStatsLastExpireTimestampForSubscription";
@Cleanup
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topic)
.create();
for (int i = 0; i < 10; i++) {
producer.send(new byte[1024 * i * 5]);
}
@Cleanup
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic(topic)
.subscriptionName("sub-1")
Expand Down Expand Up @@ -3169,10 +3227,12 @@ public void testCreateAndDeleteNamespaceWithBundles() throws Exception {
@Test
public void testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages() throws Exception {
final String topic = "persistent://prop-xyz/ns1/testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages";
@Cleanup
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic(topic)
.subscriptionName("sub-1")
.subscribe();
@Cleanup
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topic)
.create();
Expand Down Expand Up @@ -3208,6 +3268,7 @@ public void testGetTtlDurationDefaultInSeconds() throws Exception {
public void testGetReadPositionWhenJoining() throws Exception {
final String topic = "persistent://prop-xyz/ns1/testGetReadPositionWhenJoining-" + UUID.randomUUID().toString();
final String subName = "my-sub";
@Cleanup
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topic)
.enableBatching(false)
Expand All @@ -3219,12 +3280,14 @@ public void testGetReadPositionWhenJoining() throws Exception {
messageId = (MessageIdImpl) producer.send(("Hello Pulsar - " + i).getBytes());
}

List<Consumer<byte[]>> consumers = new ArrayList<>();
for (int i = 0; i < 2; i++) {
pulsarClient.newConsumer()
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic(topic)
.subscriptionType(SubscriptionType.Key_Shared)
.subscriptionName(subName)
.subscribe();
consumers.add(consumer);
}

TopicStats stats = admin.topics().getStats(topic);
Expand All @@ -3235,6 +3298,10 @@ public void testGetReadPositionWhenJoining() throws Exception {
ConsumerStats consumerStats = subStats.getConsumers().get(0);
Assert.assertEquals(consumerStats.getReadPositionWhenJoining(),
PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId() + 1).toString());

for (Consumer<byte[]> consumer : consumers) {
consumer.close();
}
}

@Test
Expand All @@ -3247,13 +3314,15 @@ public void testPartitionedTopicMsgDelayedAggregated() throws Exception {
admin.topics().createPartitionedTopic(topic, numPartitions);

for (int i = 0; i < 2; i++) {
pulsarClient.newConsumer()
@Cleanup
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic(topic)
.subscriptionType(SubscriptionType.Shared)
.subscriptionName(subName)
.subscribe();
}

@Cleanup
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(topic)
.enableBatching(false)
Expand Down