Skip to content

[Tests] Cleanup PulsarClient instances #10197

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

Merged
merlimat merged 4 commits into from
Apr 15, 2021
Merged

[Tests] Cleanup PulsarClient instances #10197

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
870d2ac
Cleanup PulsarClient instances
lhotari Apr 14, 2021
421d87a
Fix flaky PulsarMultiHostClientTest
lhotari Apr 14, 2021
8227747
Add more PulsarClient cleanups
lhotari Apr 14, 2021
fbccc7b
Fix cleanup for PersistentTopicE2ETest
lhotari Apr 15, 2021
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
3 changes: 2 additions & 1 deletion ...sl/src/test/java/org/apache/pulsar/broker/authentication/ProxySaslAuthenticationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@

import javax.security.auth.login.Configuration;

import lombok.Cleanup;
import org.apache.commons.io.FileUtils;
import org.apache.curator.shaded.com.google.common.collect.Maps;
import org.apache.pulsar.client.admin.PulsarAdmin;
Expand Down Expand Up @@ -249,6 +250,7 @@ void testAuthentication() throws Exception {
log.info("1 proxy service started {}", proxyService);

// Step 3: Pass correct client params
@Cleanup
PulsarClient proxyClient = createProxyClient(proxyServiceUrl, 1);
log.info("2 create proxy client {}, {}", proxyServiceUrl, proxyClient);

Expand Down Expand Up @@ -277,7 +279,6 @@ void testAuthentication() throws Exception {
consumer.acknowledgeCumulative(msg);
consumer.close();

proxyClient.close();
proxyService.close();
}

Expand Down
5 changes: 5 additions & 0 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 @@ -786,6 +786,7 @@ public void persistentTopics(String topicName) throws Exception {
Lists.newArrayList("persistent://prop-xyz/ns1/" + topicName));

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -891,6 +892,7 @@ public void partitionedTopics(String topicName) throws Exception {
assertEquals(subscriptions.size(), 0);

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -1542,6 +1544,7 @@ public void testDeleteSubscription() throws Exception {
assertEquals(admin.topics().getSubscriptions(persistentTopicName), Lists.newArrayList(subName));

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -2090,6 +2093,7 @@ public void persistentTopicsInvalidCursorReset() throws Exception {
assertEquals(admin.topics().getList("prop-xyz/ns1"), Lists.newArrayList(topicName));

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -2284,6 +2288,7 @@ public void testPersistentTopicExpireMessageOnPartitionTopic() throws Exception

// create consumer and subscription
URL pulsarUrl = new URL(pulsar.getWebServiceAddress());
@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(pulsarUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer = client.newConsumer().topic("persistent://prop-xyz/ns1/ds1")
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1737,6 +1737,7 @@ public void testMaxSubPerTopicPriority() throws Exception {
admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
final String topic = "persistent://" + myNamespace + "/testMaxSubPerTopic";
//Create a client that can fail quickly
@Cleanup
PulsarClient client = PulsarClient.builder().operationTimeout(2,TimeUnit.SECONDS)
.serviceUrl(brokerUrl.toString()).build();
//We can only create 2 consumers
Expand Down Expand Up @@ -1776,7 +1777,6 @@ public void testMaxSubPerTopicPriority() throws Exception {
for (Consumer<?> c : consumerList) {
c.close();
}
client.close();
}

@Test
Expand Down
6 changes: 4 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -710,6 +710,7 @@ public void persistentTopics(String topicName) throws Exception {
Lists.newArrayList("persistent://prop-xyz/use/ns1/" + topicName));

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -796,6 +797,7 @@ public void partitionedTopics(String topicName) throws Exception {
0);

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -1639,6 +1641,7 @@ public void persistentTopicsInvalidCursorReset() throws Exception {
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), Lists.newArrayList(topicName));

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -1776,6 +1779,7 @@ public void testPersistentTopicExpireMessageOnPartitionTopic() throws Exception
admin.topics().createPartitionedTopic("persistent://prop-xyz/use/ns1/ds1", 4);

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -1816,8 +1820,6 @@ public void testPersistentTopicExpireMessageOnPartitionTopic() throws Exception

producer.close();
consumer.close();
client.close();

}

/**
Expand Down
1 change: 1 addition & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,7 @@ public void nonPersistentTopics() throws Exception {
publishMessagesOnTopic("non-persistent://prop-xyz/use/ns1/" + topicName, 0, 0);

// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down
3 changes: 2 additions & 1 deletion ...er/src/test/java/org/apache/pulsar/broker/loadbalance/AntiAffinityNamespaceGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import lombok.Cleanup;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
Expand Down Expand Up @@ -518,6 +519,7 @@ public void testLoadSheddingWithAntiAffinityNamespace() throws Exception {
admin1.namespaces().setNamespaceAntiAffinityGroup(ns, namespaceAntiAffinityGroup);
}

@Cleanup
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsar1.getSafeWebServiceAddress()).build();
Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://" + namespace + "0/my-topic1")
.create();
Expand All @@ -529,7 +531,6 @@ public void testLoadSheddingWithAntiAffinityNamespace() throws Exception {

assertTrue(loadManager.shouldAntiAffinityNamespaceUnload(namespace + "0", bundle, primaryHost));
producer.close();
pulsarClient.close();
}

private boolean isLoadManagerUpdatedDomainCache(ModularLoadManagerImpl loadManager) throws Exception {
Expand Down
3 changes: 2 additions & 1 deletion ...-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LeaderElectionServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
Expand Down Expand Up @@ -80,6 +81,7 @@ public void anErrorShouldBeThrowBeforeLeaderElected() throws PulsarServerExcepti
adminClient.clusters().createCluster(clusterName, new ClusterData("http://localhost:8080"));
adminClient.tenants().createTenant(tenant, new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet(clusterName)));
adminClient.namespaces().createNamespace(tenant + "/" + namespace, 16);
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650")
.startingBackoffInterval(1, TimeUnit.MILLISECONDS)
Expand All @@ -101,7 +103,6 @@ public void anErrorShouldBeThrowBeforeLeaderElected() throws PulsarServerExcepti
.topic("persistent://" + tenant + "/" + namespace + "/1p")
.create();
producer.getTopic();
pulsar.close();

}

Expand Down
27 changes: 14 additions & 13 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

import com.beust.jcommander.internal.Maps;
import com.google.common.collect.Sets;
import lombok.Cleanup;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.client.admin.PulsarAdmin;
Expand Down Expand Up @@ -274,6 +275,7 @@ public void testConsumerBacklogEviction() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/ns-quota",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction));
@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
.build();

Expand All @@ -297,7 +299,6 @@ public void testConsumerBacklogEviction() throws Exception {

TopicStats stats = admin.topics().getStats(topic1);
assertTrue(stats.backlogSize < 10 * 1024, "Storage size is [" + stats.storageSize + "]");
client.close();
}

@Test
Expand All @@ -306,6 +307,7 @@ public void testConsumerBacklogEvictionWithAck() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/ns-quota",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction));
@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString()).build();

final String topic1 = "persistent://prop/ns-quota/topic11";
Expand All @@ -329,7 +331,6 @@ public void testConsumerBacklogEvictionWithAck() throws Exception {

TopicStats stats = admin.topics().getStats(topic1);
assertTrue(stats.backlogSize <= 10 * 1024, "Storage size is [" + stats.storageSize + "]");
client.close();
}

@Test
Expand All @@ -347,8 +348,10 @@ public void testConcurrentAckAndEviction() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
final CountDownLatch counter = new CountDownLatch(2);
final AtomicBoolean gotException = new AtomicBoolean(false);
@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
.build();
@Cleanup
PulsarClient client2 = PulsarClient.builder().serviceUrl(adminUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer1 = client2.newConsumer().topic(topic1).subscriptionName(subName1).subscribe();
Expand Down Expand Up @@ -401,8 +404,6 @@ public void run() {

TopicStats stats = admin.topics().getStats(topic1);
assertTrue(stats.backlogSize <= 10 * 1024, "Storage size is [" + stats.storageSize + "]");
client.close();
client2.close();
}

@Test
Expand All @@ -420,11 +421,13 @@ public void testNoEviction() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(2);
final CountDownLatch counter = new CountDownLatch(2);
final AtomicBoolean gotException = new AtomicBoolean(false);
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();

final Consumer<byte[]> consumer1 = client.newConsumer().topic(topic1).subscriptionName(subName1).subscribe();
final Consumer<byte[]> consumer2 = client.newConsumer().topic(topic1).subscriptionName(subName2).subscribe();
@Cleanup
final PulsarClient client2 = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();

Expand Down Expand Up @@ -467,8 +470,6 @@ public void run() {
ConsumerThread.start();
counter.await();
assertFalse(gotException.get());
client.close();
client2.close();
}

@Test
Expand All @@ -486,13 +487,16 @@ public void testEvictionMulti() throws Exception {
final CyclicBarrier barrier = new CyclicBarrier(4);
final CountDownLatch counter = new CountDownLatch(4);
final AtomicBoolean gotException = new AtomicBoolean(false);
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();

final Consumer<byte[]> consumer1 = client.newConsumer().topic(topic1).subscriptionName(subName1).subscribe();
final Consumer<byte[]> consumer2 = client.newConsumer().topic(topic1).subscriptionName(subName2).subscribe();
@Cleanup
final PulsarClient client3 = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();
@Cleanup
final PulsarClient client2 = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();

Expand Down Expand Up @@ -575,9 +579,6 @@ public void run() {

TopicStats stats = admin.topics().getStats(topic1);
assertTrue(stats.backlogSize <= 15 * 1024, "Storage size is [" + stats.storageSize + "]");
client.close();
client2.close();
client3.close();
}

@Test
Expand All @@ -586,6 +587,7 @@ public void testAheadProducerOnHold() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/quotahold",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.producer_request_hold));
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();
final String topic1 = "persistent://prop/quotahold/hold";
Expand Down Expand Up @@ -616,7 +618,6 @@ public void testAheadProducerOnHold() throws Exception {
TopicStats stats = admin.topics().getStats(topic1);
assertEquals(stats.publishers.size(), 0,
"Number of producers on topic " + topic1 + " are [" + stats.publishers.size() + "]");
client.close();
}

@Test
Expand All @@ -625,6 +626,7 @@ public void testAheadProducerOnHoldTimeout() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/quotahold",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.producer_request_hold));
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();
final String topic1 = "persistent://prop/quotahold/holdtimeout";
Expand All @@ -651,7 +653,6 @@ public void testAheadProducerOnHoldTimeout() throws Exception {
}

assertTrue(gotException, "timeout did not occur");
client.close();
}

@Test
Expand All @@ -660,6 +661,7 @@ public void testProducerException() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/quotahold",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.producer_exception));
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();
final String topic1 = "persistent://prop/quotahold/except";
Expand Down Expand Up @@ -688,7 +690,6 @@ public void testProducerException() throws Exception {
}

assertTrue(gotException, "backlog exceeded exception did not occur");
client.close();
}

@Test
Expand All @@ -697,6 +698,7 @@ public void testProducerExceptionAndThenUnblock() throws Exception {
Maps.newHashMap());
admin.namespaces().setBacklogQuota("prop/quotahold",
new BacklogQuota(10 * 1024, BacklogQuota.RetentionPolicy.producer_exception));
@Cleanup
final PulsarClient client = PulsarClient.builder().serviceUrl(adminUrl.toString())
.statsInterval(0, TimeUnit.SECONDS).build();
final String topic1 = "persistent://prop/quotahold/exceptandunblock";
Expand Down Expand Up @@ -747,7 +749,6 @@ public void testProducerExceptionAndThenUnblock() throws Exception {
sendException = e;
}
assertFalse(gotException, "unable to publish due to " + sendException);
client.close();
}

private static final Logger LOG = LoggerFactory.getLogger(BacklogQuotaManagerTest.class);
Expand Down
7 changes: 3 additions & 4 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ public BrokerBkEnsemblesTests(int numberOfBookies) {
public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {

ZooKeeper zk = bkEnsemble.getZkClient();
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -153,8 +154,6 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {

producer.close();
consumer.close();
client.close();

}

/**
Expand All @@ -176,6 +175,7 @@ public void testSkipCorruptDataLedger() throws Exception {
// Ensure intended state for autoSkipNonRecoverableData
admin.brokers().updateDynamicConfiguration("autoSkipNonRecoverableData", "false");

@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand Down Expand Up @@ -274,11 +274,11 @@ public void testSkipCorruptDataLedger() throws Exception {

producer.close();
consumer.close();
client.close();
}

@Test(timeOut = 20000)
public void testTopicWithWildCardChar() throws Exception {
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
Expand All @@ -303,7 +303,6 @@ public void testTopicWithWildCardChar() throws Exception {
Assert.assertEquals(msg.getData(), content);
consumer.close();
producer.close();
client.close();
}


Expand Down
Loading