[fix][client] Prevent client shutdown from leaking event loop threads when DNS resolver close fails

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -1071,11 +1071,21 @@ public void shutdown() throws PulsarClientException {
             }
 
             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 {

pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java

@@ -24,6 +24,7 @@
 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.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 void testInitializeWithoutTimer() throws Exception {
         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();