[fix][client] Fix waiting lookup requests not dispatched on timeout in ClientCnx

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

@@ -1008,8 +1008,7 @@ public CompletableFuture<LookupDataResult> 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))) {

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<BinaryProtoLookupService.LookupDataResult> 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<BinaryProtoLookupService.LookupDataResult> firstFuture =
+                cnx.newLookup(null, 1L);
+
+        // Second lookup should go into the waiting queue
+        CompletableFuture<BinaryProtoLookupService.LookupDataResult> 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<ClientCnx> test) {
         ThreadFactory threadFactory = new DefaultThreadFactory(testName);
         EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);