(pulsar) branch master updated: [fix][test] Fix resource leak when PulsarTestClient is closed or shutdown (#21468)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari 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 bb8082ebc60 [fix][test] Fix resource leak when PulsarTestClient is closed or shutdown (#21468)
bb8082ebc60 is described below

commit bb8082ebc60832e70828890cdb4016d66dcce2a8
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 30 10:32:21 2023 +0200

    [fix][test] Fix resource leak when PulsarTestClient is closed or shutdown (#21468)
---
 .../pulsar/client/impl/PulsarTestClient.java       | 32 +++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
index 6555c152bac..d588ac8626f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
@@ -24,9 +24,11 @@ import io.netty.util.concurrent.DefaultThreadFactory;
 import java.io.IOException;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -50,6 +52,7 @@ import org.awaitility.Awaitility;
  *   called after the message to send out has been added to the pending messages in the client.
  *
  */
+@Slf4j
 public class PulsarTestClient extends PulsarClientImpl {
     private volatile int overrideRemoteEndpointProtocolVersion;
     private volatile boolean rejectNewConnections;
@@ -73,7 +76,7 @@ public class PulsarTestClient extends PulsarClientImpl {
         // method.
         EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(clientConfigurationData.getNumIoThreads(),
                 false,
-                new DefaultThreadFactory("pulsar-client-io", Thread.currentThread().isDaemon()));
+                new DefaultThreadFactory("pulsar-test-client-io", Thread.currentThread().isDaemon()));
 
         AtomicReference<Supplier<ClientCnx>> clientCnxSupplierReference = new AtomicReference<>();
         ConnectionPool connectionPool = new ConnectionPool(clientConfigurationData, eventLoopGroup,
@@ -217,4 +220,31 @@ public class PulsarTestClient extends PulsarClientImpl {
     public void dropOpSendMessages() {
         this.dropOpSendMessages = true;
     }
+
+    @Override
+    public CompletableFuture<Void> closeAsync() {
+        return super.closeAsync().handle((__, t) -> {
+            shutdownCnxPoolAndEventLoopGroup();
+            return null;
+        });
+    }
+
+    @Override
+    public void shutdown() throws PulsarClientException {
+        super.shutdown();
+        shutdownCnxPoolAndEventLoopGroup();
+    }
+
+    private void shutdownCnxPoolAndEventLoopGroup() {
+        try {
+            getCnxPool().close();
+        } catch (Exception e) {
+            log.warn("Error closing connection pool", e);
+        }
+        try {
+            eventLoopGroup.shutdownGracefully().get(5, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            log.warn("Error closing event loop group", e);
+        }
+    }
 }