This is an automated email from the ASF dual-hosted git repository.
tison 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 5e16ecf5007 [fix][test] Fix resource leaks in LoadBalancerTest (#21443)
5e16ecf5007 is described below
commit 5e16ecf50072e5c24bacb94a3c5c3265cd3cc9e4
Author: Lari Hotari <[email protected]>
AuthorDate: Thu Oct 26 03:44:59 2023 +0300
[fix][test] Fix resource leaks in LoadBalancerTest (#21443)
---
.../broker/loadbalance/LoadBalancerTest.java | 28 ++++++----------------
1 file changed, 7 insertions(+), 21 deletions(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
index 7cc4499df97..50afb71ea09 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
@@ -34,10 +34,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import lombok.SneakyThrows;
@@ -88,8 +85,6 @@ import org.testng.annotations.Test;
public class LoadBalancerTest {
LocalBookkeeperEnsemble bkEnsemble;
- ExecutorService executor = new ThreadPoolExecutor(5, 20, 30,
TimeUnit.SECONDS, new LinkedBlockingQueue<>());
-
private static final Logger log =
LoggerFactory.getLogger(LoadBalancerTest.class);
private static final int MAX_RETRIES = 15;
@@ -150,13 +145,10 @@ public class LoadBalancerTest {
@AfterMethod(alwaysRun = true)
void shutdown() throws Exception {
log.info("--- Shutting down ---");
- executor.shutdownNow();
for (int i = 0; i < BROKER_COUNT; i++) {
pulsarAdmins[i].close();
- if (pulsarServices[i] != null) {
- pulsarServices[i].close();
- }
+ pulsarServices[i].close();
}
bkEnsemble.stop();
@@ -670,27 +662,21 @@ public class LoadBalancerTest {
*/
@Test
public void testLeaderElection() throws Exception {
- // this.pulsarServices is the reference to all of the PulsarServices
- // it is used in order to clean up the resources
- PulsarService[] allServices = new PulsarService[pulsarServices.length];
- System.arraycopy(pulsarServices, 0, allServices, 0,
pulsarServices.length);
for (int i = 0; i < BROKER_COUNT - 1; i++) {
List<PulsarService> activePulsar = new ArrayList<>();
List<PulsarService> followerPulsar = new ArrayList<>();
LeaderBroker oldLeader = null;
PulsarService leaderPulsar = null;
for (int j = 0; j < BROKER_COUNT; j++) {
- if (allServices[j].getState() != PulsarService.State.Closed) {
- activePulsar.add(allServices[j]);
- LeaderElectionService les =
allServices[j].getLeaderElectionService();
+ PulsarService pulsarService = pulsarServices[j];
+ if (pulsarService.getState() != PulsarService.State.Closed) {
+ activePulsar.add(pulsarService);
+ LeaderElectionService les =
pulsarService.getLeaderElectionService();
if (les.isLeader()) {
oldLeader = les.getCurrentLeader().get();
- leaderPulsar = allServices[j];
- // set the refence to null in the main array,
- // in order to prevent closing this PulsarService twice
- pulsarServices[i] = null;
+ leaderPulsar = pulsarService;
} else {
- followerPulsar.add(allServices[j]);
+ followerPulsar.add(pulsarService);
}
}
}
