(pulsar) branch master updated: [fix][client] Prevent client shutdown from leaking event loop threads when DNS resolver close fails (#26045)

[mm...@apache.org]

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

merlimat 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 d58727f3bf2 [fix][client] Prevent client shutdown from leaking event loop threads when DNS resolver close fails (#26045)
d58727f3bf2 is described below

commit d58727f3bf2777170ccc19afccf0c4364b94e4da
Author: David Kjerrumgaard <35...@users.noreply.github.com>
AuthorDate: Wed Jun 17 12:00:56 2026 -0700

    [fix][client] Prevent client shutdown from leaking event loop threads when DNS resolver close fails (#26045)
---
 .../pulsar/client/impl/PulsarClientImpl.java       | 14 +++++++--
 .../pulsar/client/impl/PulsarClientImplTest.java   | 34 ++++++++++++++++++++++
 2 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index 23e6b94dcea..9d0a814c5eb 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -1071,11 +1071,21 @@ public class PulsarClientImpl implements PulsarClient {
             }
 
             if (addressResolver != null) {
-                addressResolver.close();
+                try {
+                    addressResolver.close();
+                } catch (Throwable t) {
+                    log.warn().exception(t).log("Failed to close addressResolver");
+                    throwable = t;
+                }
             }
 
             if (dnsResolverGroupLocalInstance != null) {
-                dnsResolverGroupLocalInstance.close();
+                try {
+                    dnsResolverGroupLocalInstance.close();
+                } catch (Throwable t) {
+                    log.warn().exception(t).log("Failed to close dnsResolverGroup");
+                    throwable = t;
+                }
             }
 
             try {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
index b742af7336d..702410f9612 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
@@ -24,6 +24,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -39,6 +40,7 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoopGroup;
+import io.netty.resolver.AddressResolver;
 import io.netty.resolver.dns.DefaultDnsServerAddressStreamProvider;
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.concurrent.DefaultThreadFactory;
@@ -183,6 +185,38 @@ public class PulsarClientImplTest {
         verify(timer).stop();
     }
 
+    @Test
+    public void testShutdownContinuesWhenAddressResolverCloseFails() throws Exception {
+        ClientConfigurationData conf = new ClientConfigurationData();
+        conf.setServiceUrl("pulsar://localhost:6650");
+        PulsarClientImpl client = new PulsarClientImpl(conf);
+
+        // Simulate the Netty DNS resolver shutdown race where AddressResolver.close()
+        // throws "channel not registered to an event loop".
+        @SuppressWarnings("unchecked")
+        AddressResolver<InetSocketAddress> failingResolver = mock(AddressResolver.class);
+        doThrow(new IllegalStateException("channel not registered to an event loop"))
+                .when(failingResolver).close();
+        Field addressResolverField = PulsarClientImpl.class.getDeclaredField("addressResolver");
+        addressResolverField.setAccessible(true);
+        addressResolverField.set(client, failingResolver);
+
+        // Replace the timer with a mock so we can assert that shutdown still reaches the
+        // cleanup steps that run after the address resolver is closed.
+        HashedWheelTimer timer = mock(HashedWheelTimer.class);
+        Field timerField = PulsarClientImpl.class.getDeclaredField("timer");
+        timerField.setAccessible(true);
+        timerField.set(client, timer);
+
+        // shutdown() still surfaces the original failure, but it must not abort early:
+        // the remaining resources have to be released regardless. The key assertion is
+        // that timer.stop() (a step that runs after the resolver is closed) is reached.
+        assertThrows(IllegalStateException.class, client::shutdown);
+
+        verify(failingResolver).close();
+        verify(timer).stop();
+    }
+
     @Test
     public void testInitializeWithTimer() throws PulsarClientException {
         ClientConfigurationData conf = new ClientConfigurationData();