From a65f080f4bbee5813aefee7e2a8e43a65743297b Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 17 Jun 2026 09:22:15 -0700 Subject: [PATCH 1/2] [fix][client] Prevent client shutdown from leaking event loop threads when DNS resolver close fails PulsarClientImpl.shutdown() closed the Netty addressResolver and the local DnsResolverGroup instance without a try/catch, unlike every other resource it closes. When addressResolver.close() throws (e.g. the Netty 'IllegalStateException: channel not registered to an event loop' race that can happen when a DNS lookup is in flight at shutdown), the exception aborts the rest of shutdown() before shutdownEventLoopGroup(), closeCnxPool(), timer.stop() and shutdownExecutors() run. The eventLoopGroup's non-daemon threads are then leaked and the JVM never exits. Wrap both DNS resolver close calls in try/catch and record the throwable, so shutdown always proceeds to release the remaining resources. --- .../pulsar/client/impl/PulsarClientImpl.java | 14 ++++++++++++-- 1 file changed, 12 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 23e6b94dcea99..9d0a814c5eb1e 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 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 { From cf0904160c26db49e9a15a9047f7973074d14df3 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 17 Jun 2026 10:23:27 -0700 Subject: [PATCH 2/2] [fix][client] Add regression test for shutdown continuing past DNS resolver close failure Verifies that PulsarClientImpl.shutdown() still releases the remaining resources (asserted via timer.stop()) when addressResolver.close() throws, simulating the Netty 'channel not registered to an event loop' race. The test fails without the shutdown fix (timer.stop() is never reached) and passes with it. --- .../client/impl/PulsarClientImplTest.java | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) 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 b742af7336d1e..702410f961238 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.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 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();