From 515e149caeaf3473775de9abe070ddcc77d78835 Mon Sep 17 00:00:00 2001 From: ninjazhou <843520313@qq.com> Date: Thu, 2 Jul 2026 22:04:19 +0800 Subject: [PATCH] [fix][client] Fix waiting lookup requests not dispatched on timeout in ClientCnx --- .../apache/pulsar/client/impl/ClientCnx.java | 11 +- .../pulsar/client/impl/ClientCnxTest.java | 137 ++++++++++++++++++ 2 files changed, 145 insertions(+), 3 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index 16d78585531a6..6db2f337b8380 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -1008,8 +1008,7 @@ public CompletableFuture newLookup(ByteBuf request, long reque if (pendingLookupRequestSemaphore.tryAcquire()) { future.whenComplete((lookupDataResult, throwable) -> { if (throwable instanceof ConnectException - || throwable instanceof PulsarClientException.LookupException - || FutureUtil.unwrapCompletionException(throwable) instanceof TimeoutException) { + || throwable instanceof PulsarClientException.LookupException) { pendingLookupRequestSemaphore.release(); } }); @@ -1796,7 +1795,13 @@ private void checkRequestTimeout() { TimedCompletableFuture requestFuture = pendingRequests.get(request.requestId); if (requestFuture != null && !requestFuture.hasGotResponse()) { - pendingRequests.remove(request.requestId, requestFuture); + if (request.requestType == RequestType.Lookup) { + // For Lookup type, use getAndRemovePendingLookupRequest to release the semaphore + // and drive the waiting queue + getAndRemovePendingLookupRequest(request.requestId); + } else { + pendingRequests.remove(request.requestId, requestFuture); + } if (!requestFuture.isDone()) { String timeoutMessage = request.requestType.getDescription() + " timeout"; if (requestFuture.completeExceptionally(new TimeoutException(timeoutMessage))) { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java index db1725089802c..d5df01339364d 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java @@ -37,6 +37,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.PulsarClientException.BrokerMetadataException; @@ -373,6 +374,142 @@ public void testUpdateWatcher() { }); } + /** + * Test that when a lookup request times out, the semaphore is properly released + * so that subsequent lookup requests can still be sent. + */ + @Test + public void testLookupTimeoutReleasesSemaphore() throws Exception { + EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, + new DefaultThreadFactory("testLookupTimeoutReleasesSemaphore")); + ClientConfigurationData conf = new ClientConfigurationData(); + conf.setOperationTimeoutMs(10); + conf.setKeepAliveIntervalSeconds(0); + ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop); + ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext(); + cnx.channelActive(ctx); + + int initialPermits = cnx.getPendingLookupRequestSemaphore().availablePermits(); + + // Send a lookup request that will time out + CompletableFuture future = + cnx.newLookup(null, 1L); + + // Wait for the timeout to trigger + try { + future.get(2, TimeUnit.SECONDS); + fail("Should have timed out"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof PulsarClientException.TimeoutException); + } + + // Verify semaphore is released back to initial permits + Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), initialPermits); + }); + + eventLoop.shutdownGracefully(); + } + + /** + * Test that when a lookup request times out, waiting lookup requests in the queue + * are properly dispatched (the waiting queue is driven). + */ + @Test + public void testLookupTimeoutDrivesWaitingQueue() throws Exception { + EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, + new DefaultThreadFactory("testLookupTimeoutDrivesWaitingQueue")); + ClientConfigurationData conf = new ClientConfigurationData(); + // concurrentLookupRequest=50 by default, maxLookupRequest=50000 by default + conf.setOperationTimeoutMs(50); + conf.setKeepAliveIntervalSeconds(0); + conf.setConcurrentLookupRequest(1); // Only allow 1 concurrent lookup + conf.setMaxLookupRequest(10); // Allow up to 10 total (9 waiting) + ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop); + ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext(); + cnx.channelActive(ctx); + cnx.state = ClientCnx.State.Ready; + + // First lookup occupies the only concurrent slot + CompletableFuture firstFuture = + cnx.newLookup(null, 1L); + + // Second lookup should go into the waiting queue + CompletableFuture secondFuture = + cnx.newLookup(null, 2L); + + // The second future should not be completed yet (it's waiting) + assertFalse(secondFuture.isDone()); + + // Wait for the first request to time out - this should drive the waiting queue + // and dispatch the second request + try { + firstFuture.get(2, TimeUnit.SECONDS); + fail("First future should have timed out"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof PulsarClientException.TimeoutException); + } + + // After the first request times out, the second request should have been + // dispatched from the waiting queue (it will also eventually time out) + try { + secondFuture.get(2, TimeUnit.SECONDS); + fail("Second future should have timed out"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof PulsarClientException.TimeoutException); + } + + // Verify all semaphores are properly released + Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), 1); + }); + + eventLoop.shutdownGracefully(); + } + + /** + * Test that when multiple lookup requests time out, all waiting requests in the queue + * are eventually dispatched and no semaphore leak occurs. + */ + @Test + public void testMultipleLookupTimeoutsNoSemaphoreLeak() throws Exception { + EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, + new DefaultThreadFactory("testMultipleLookupTimeoutsNoSemaphoreLeak")); + ClientConfigurationData conf = new ClientConfigurationData(); + conf.setOperationTimeoutMs(30); + conf.setKeepAliveIntervalSeconds(0); + conf.setConcurrentLookupRequest(2); // Allow 2 concurrent lookups + conf.setMaxLookupRequest(10); // Allow up to 10 total (8 waiting) + ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop); + ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext(); + cnx.channelActive(ctx); + cnx.state = ClientCnx.State.Ready; + + // Send 5 lookup requests: 2 will be concurrent, 3 will be in waiting queue + CompletableFuture[] futures = new CompletableFuture[5]; + for (int i = 0; i < 5; i++) { + futures[i] = cnx.newLookup(null, i + 1L); + } + + // Wait for all futures to complete (all should time out eventually) + for (int i = 0; i < 5; i++) { + try { + futures[i].get(5, TimeUnit.SECONDS); + fail("Future " + i + " should have timed out"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof PulsarClientException.TimeoutException, + "Future " + i + " should fail with TimeoutException but got: " + e.getCause()); + } + } + + // Verify all semaphore permits are released (no leak) + Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), 2); + }); + + eventLoop.shutdownGracefully(); + } + private void withConnection(String testName, Consumer test) { ThreadFactory threadFactory = new DefaultThreadFactory(testName); EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);