From ea4c3cd7e95f67ac6aa55d56f339e4325054d257 Mon Sep 17 00:00:00 2001 From: congbo Date: Thu, 4 Dec 2025 20:20:28 +0800 Subject: [PATCH 1/2] [fix][client] Fix lookup request semaphore not release by timeoutException --- .../pulsar/client/impl/LookupRetryTest.java | 34 +++++++++++++++++++ .../apache/pulsar/client/impl/ClientCnx.java | 3 +- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java index 7796671307196..1142ad233a6d1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java @@ -20,9 +20,13 @@ import static org.apache.pulsar.common.protocol.Commands.newLookupErrorResponse; import static org.apache.pulsar.common.protocol.Commands.newPartitionMetadataResponse; +import static org.testng.AssertJUnit.fail; + import com.google.common.collect.Sets; import java.util.Queue; +import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -34,13 +38,16 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.common.api.proto.CommandLookupTopic; import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadata; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.TenantInfoImpl; +import org.apache.pulsar.common.util.FutureUtil; import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -51,6 +58,7 @@ public class LookupRetryTest extends MockedPulsarServiceBaseTest { private static final String subscription = "reader-sub"; private final AtomicInteger connectionsCreated = new AtomicInteger(0); private final ConcurrentHashMap> failureMap = new ConcurrentHashMap<>(); + private static final String TEST_TIME_OUT_SEMAPHORE_RELEASE = "testTimeoutReleasePendingLookupRequestSemaphore"; @BeforeClass @Override @@ -110,6 +118,29 @@ public void testGetPartitionedMetadataRetries() throws Exception { } } + @Test + public void testTimeoutReleasePendingLookupRequestSemaphore() throws Exception { + PulsarClientImpl client = (PulsarClientImpl) newClient(); + + LookupService lookup = client.getLookup(); + + CompletableFuture future = + lookup.getPartitionedTopicMetadata(TopicName.get(TEST_TIME_OUT_SEMAPHORE_RELEASE), false); + try { + future.get(); + fail(); + } catch (Exception e) { + Assert.assertTrue(FutureUtil.unwrapCompletionException(e) instanceof PulsarClientException.TimeoutException); + } + + Set> clientCnxs = client.getCnxPool().getConnections(); + Assert.assertEquals(clientCnxs.size(), 1); + ClientCnx clientCnx = ((CompletableFuture) clientCnxs.toArray()[0]).get(); + Assert.assertEquals(clientCnx.getPendingLookupRequestSemaphore().availablePermits(), + client.conf.getConcurrentLookupRequest()); + client.close(); + } + @Test public void testTimeoutRetriesOnPartitionMetadata() throws Exception { try (PulsarClient client = newClient(); @@ -282,6 +313,9 @@ private Queue errorList(String topicName) { @Override protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadata) { TopicName t = TopicName.get(partitionMetadata.getTopic()); + if (t.getLocalName().equals(TEST_TIME_OUT_SEMAPHORE_RELEASE)) { + return; + } LookupError error = errorList(t.getLocalName()).poll(); if (error == LookupError.TOO_MANY) { final long requestId = partitionMetadata.getRequestId(); 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 07df92f501e29..184b210e2b0cd 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 @@ -922,7 +922,8 @@ public CompletableFuture newLookup(ByteBuf request, long reque if (pendingLookupRequestSemaphore.tryAcquire()) { future.whenComplete((lookupDataResult, throwable) -> { if (throwable instanceof ConnectException - || throwable instanceof PulsarClientException.LookupException) { + || throwable instanceof PulsarClientException.LookupException + || FutureUtil.unwrapCompletionException(throwable) instanceof TimeoutException) { pendingLookupRequestSemaphore.release(); } }); From f8fca27b68755156daa2204db8f57f296f29ee79 Mon Sep 17 00:00:00 2001 From: congbo Date: Thu, 4 Dec 2025 20:38:25 +0800 Subject: [PATCH 2/2] fix code style --- .../java/org/apache/pulsar/client/impl/LookupRetryTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java index 1142ad233a6d1..f8ebebd53e43a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/LookupRetryTest.java @@ -21,7 +21,6 @@ import static org.apache.pulsar.common.protocol.Commands.newLookupErrorResponse; import static org.apache.pulsar.common.protocol.Commands.newPartitionMetadataResponse; import static org.testng.AssertJUnit.fail; - import com.google.common.collect.Sets; import java.util.Queue; import java.util.Set; @@ -130,7 +129,8 @@ public void testTimeoutReleasePendingLookupRequestSemaphore() throws Exception { future.get(); fail(); } catch (Exception e) { - Assert.assertTrue(FutureUtil.unwrapCompletionException(e) instanceof PulsarClientException.TimeoutException); + Assert.assertTrue(FutureUtil.unwrapCompletionException(e) + instanceof PulsarClientException.TimeoutException); } Set> clientCnxs = client.getCnxPool().getConnections();