You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ha...@apache.org on 2026/07/15 12:48:03 UTC
(pulsar) branch master updated: [fix][client] Fix lookup permit double-release, waiting queue starvation and timeout-response races in ClientCnx (#26143)
This is an automated email from the ASF dual-hosted git repository.
hanmz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 8576283da4ad [fix][client] Fix lookup permit double-release, waiting queue starvation and timeout-response races in ClientCnx (#26143)
8576283da4ad is described below
commit 8576283da4ad3f3742168fc4bf4a527f31923297
Author: zhou zhuohan <84...@qq.com>
AuthorDate: Wed Jul 15 20:47:57 2026 +0800
[fix][client] Fix lookup permit double-release, waiting queue starvation and timeout-response races in ClientCnx (#26143)
---
.../org/apache/pulsar/client/impl/ClientCnx.java | 176 +++++++-----
.../apache/pulsar/client/impl/ClientCnxTest.java | 298 +++++++++++++++++++++
2 files changed, 400 insertions(+), 74 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 16d78585531a..14129a05d177 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
@@ -41,7 +41,9 @@ import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
@@ -134,6 +136,11 @@ public class ClientCnx extends PulsarHandler {
.expectedItems(16)
.concurrencyLevel(1)
.build();
+ // pendingRequests stores all pending request futures but does not preserve the command type,
+ // so there is no way to distinguish lookup requests from other requests (e.g. producer/consumer creation,
+ // getTopics, getLastMessageId). This set tracks which requestIds belong to lookup requests,
+ // so that removePendingRequest can release the lookup semaphore correctly.
+ private final Set<Long> pendingLookupRequestIds = ConcurrentHashMap.newKeySet();
// LookupRequests that waiting in client side.
private final Queue<Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>>> waitingLookupRequests;
@@ -385,13 +392,16 @@ public class ClientCnx extends PulsarHandler {
ConnectException e = new ConnectException(
"Disconnected from server at " + ctx.channel().remoteAddress());
+ // Fail out all waiting lookup requests first, and clear the queue so that
+ // releasePermitAndDriveWaitingQueue won't dispatch requests on a dead connection.
+ waitingLookupRequests.forEach(pair -> pair.getRight().getRight().completeExceptionally(e));
+ waitingLookupRequests.clear();
// Fail out all the pending ops
pendingRequests.forEach((key, future) -> {
- if (pendingRequests.remove(key, future) && !future.isDone()) {
+ if (removePendingRequest(key, future) && !future.isDone()) {
future.completeExceptionally(e);
}
});
- waitingLookupRequests.forEach(pair -> pair.getRight().getRight().completeExceptionally(e));
// Notify all attached producers/consumers so they have a chance to reconnect
producers.forEach((id, producer) -> producer.connectionClosed(this, Optional.empty(), Optional.empty()));
@@ -403,8 +413,6 @@ public class ClientCnx extends PulsarHandler {
scalableTopicsWatchers.forEach((__, session) -> session.connectionClosed());
tcAssignmentsWatchers.forEach((__, session) -> session.connectionClosed());
- waitingLookupRequests.clear();
-
producers.clear();
consumers.clear();
topicListWatchers.clear();
@@ -576,8 +584,9 @@ public class ClientCnx extends PulsarHandler {
protected void handleAckResponse(CommandAckResponse ackResponse) {
checkArgument(state == State.Ready);
checkArgument(ackResponse.getRequestId() >= 0);
- CompletableFuture<?> completableFuture = pendingRequests.remove(ackResponse.getRequestId());
- if (completableFuture != null && !completableFuture.isDone()) {
+ long requestId = ackResponse.getRequestId();
+ CompletableFuture<?> completableFuture = pendingRequests.get(requestId);
+ if (completableFuture != null && removePendingRequest(requestId, completableFuture)) {
if (!ackResponse.hasError()) {
completableFuture.complete(null);
} else {
@@ -622,8 +631,8 @@ public class ClientCnx extends PulsarHandler {
log.debug().attr("requestId", success.getRequestId())
.log("Received success response from server");
long requestId = success.getRequestId();
- CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ CompletableFuture<?> requestFuture = pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.complete(null);
} else {
duplicatedResponseCounter.incrementAndGet();
@@ -639,8 +648,8 @@ public class ClientCnx extends PulsarHandler {
.log("Received success GetLastMessageId response from server");
long requestId = success.getRequestId();
CompletableFuture<CommandGetLastMessageIdResponse> requestFuture =
- (CompletableFuture<CommandGetLastMessageIdResponse>) pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ (CompletableFuture<CommandGetLastMessageIdResponse>) pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.complete(new CommandGetLastMessageIdResponse().copyFrom(success));
} else {
duplicatedResponseCounter.incrementAndGet();
@@ -672,8 +681,8 @@ public class ClientCnx extends PulsarHandler {
}
CompletableFuture<ProducerResponse> requestFuture =
- (CompletableFuture<ProducerResponse>) pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ (CompletableFuture<ProducerResponse>) pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
ProducerResponse pr = new ProducerResponse(success.getProducerName(),
success.getLastSequenceId(),
success.getSchemaVersion(),
@@ -696,9 +705,10 @@ public class ClientCnx extends PulsarHandler {
});
long requestId = lookupResult.getRequestId();
- CompletableFuture<LookupDataResult> requestFuture = getAndRemovePendingLookupRequest(requestId);
+ CompletableFuture<LookupDataResult> requestFuture =
+ (CompletableFuture<LookupDataResult>) pendingRequests.get(requestId);
- if (requestFuture != null) {
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
if (requestFuture.isCompletedExceptionally()) {
log.debug()
.attr("requestId", lookupResult.getRequestId())
@@ -723,6 +733,7 @@ public class ClientCnx extends PulsarHandler {
requestFuture.complete(new LookupDataResult(lookupResult));
}
} else {
+ duplicatedResponseCounter.incrementAndGet();
log.warn().attr("requestId", lookupResult.getRequestId())
.log("Received unknown request id from server");
}
@@ -739,9 +750,10 @@ public class ClientCnx extends PulsarHandler {
});
long requestId = lookupResult.getRequestId();
- CompletableFuture<LookupDataResult> requestFuture = getAndRemovePendingLookupRequest(requestId);
+ CompletableFuture<LookupDataResult> requestFuture =
+ (CompletableFuture<LookupDataResult>) pendingRequests.get(requestId);
- if (requestFuture != null) {
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
if (requestFuture.isCompletedExceptionally()) {
log.debug()
.attr("requestId", lookupResult.getRequestId())
@@ -766,6 +778,7 @@ public class ClientCnx extends PulsarHandler {
requestFuture.complete(new LookupDataResult(lookupResult.getPartitions()));
}
} else {
+ duplicatedResponseCounter.incrementAndGet();
log.warn().attr("requestId", lookupResult.getRequestId())
.log("Received unknown request id from server");
}
@@ -814,41 +827,63 @@ public class ClientCnx extends PulsarHandler {
// caller of this method needs to be protected under pendingLookupRequestSemaphore
private void addPendingLookupRequests(long requestId, TimedCompletableFuture<LookupDataResult> future) {
+ pendingLookupRequestIds.add(requestId);
pendingRequests.put(requestId, future);
requestTimeoutQueue.add(new RequestTime(requestId, RequestType.Lookup));
}
- private CompletableFuture<LookupDataResult> getAndRemovePendingLookupRequest(long requestId) {
- CompletableFuture<LookupDataResult> result =
- (CompletableFuture<LookupDataResult>) pendingRequests.remove(requestId);
- if (result != null) {
- Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>> firstOneWaiting =
- waitingLookupRequests.poll();
- if (firstOneWaiting != null) {
- maxLookupRequestSemaphore.release();
- // schedule a new lookup in.
- eventLoopGroup.execute(() -> {
- long newId = firstOneWaiting.getLeft();
- TimedCompletableFuture<LookupDataResult> newFuture = firstOneWaiting.getRight().getRight();
- addPendingLookupRequests(newId, newFuture);
- ctx.writeAndFlush(firstOneWaiting.getRight().getLeft()).addListener(writeFuture -> {
- if (!writeFuture.isSuccess()) {
- log.warn()
- .attr("requestId", newId)
- .exceptionMessage(writeFuture.cause())
- .log("Failed to send request to broker");
- getAndRemovePendingLookupRequest(newId);
+ /**
+ * Release the lookup semaphore permit and drive the waiting queue.
+ * This is the single centralized primitive for permit release/transfer.
+ */
+ private void releasePermitAndDriveWaitingQueue() {
+ Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>> firstOneWaiting =
+ waitingLookupRequests.poll();
+ if (firstOneWaiting != null) {
+ maxLookupRequestSemaphore.release();
+ // schedule a new lookup in.
+ eventLoopGroup.execute(() -> {
+ long newId = firstOneWaiting.getLeft();
+ TimedCompletableFuture<LookupDataResult> newFuture = firstOneWaiting.getRight().getRight();
+ addPendingLookupRequests(newId, newFuture);
+ ctx.writeAndFlush(firstOneWaiting.getRight().getLeft()).addListener(writeFuture -> {
+ if (!writeFuture.isSuccess()) {
+ log.warn()
+ .attr("requestId", newId)
+ .exceptionMessage(writeFuture.cause())
+ .log("Failed to send request to broker");
+ if (removePendingRequest(newId, newFuture)) {
newFuture.completeExceptionally(writeFuture.cause());
}
- });
+ }
});
- } else {
- pendingLookupRequestSemaphore.release();
- }
+ });
} else {
- duplicatedResponseCounter.incrementAndGet();
+ pendingLookupRequestSemaphore.release();
+ }
+ }
+
+ /**
+ * Unified cleanup primitive for all pending requests.
+ * Uses pendingRequests.remove(requestId, expectedFuture) as the single CAS contention point.
+ * Only the thread that successfully removes the entry from pendingRequests owns the cleanup
+ * responsibility (including permit release for lookup requests).
+ *
+ * @param requestId the request ID to remove
+ * @param expectedFuture the expected future value for CAS comparison
+ * @return true if this call successfully obtained ownership and performed cleanup
+ */
+ private boolean removePendingRequest(long requestId, CompletableFuture<?> expectedFuture) {
+ // CAS: only one thread can successfully remove from pendingRequests
+ if (!pendingRequests.remove(requestId, expectedFuture)) {
+ return false;
}
- return result;
+ // Won the CAS. If it's a lookup request, release the permit and drive waiting queue.
+ if (pendingLookupRequestIds.contains(requestId)) {
+ releasePermitAndDriveWaitingQueue();
+ pendingLookupRequestIds.remove(requestId);
+ }
+ return true;
}
@Override
@@ -901,8 +936,8 @@ public class ClientCnx extends PulsarHandler {
log.error().attr("message", error.getMessage()).log("Get not allowed error");
connectionFuture.completeExceptionally(new PulsarClientException.NotAllowedException(error.getMessage()));
}
- CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ CompletableFuture<?> requestFuture = pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.completeExceptionally(
getPulsarClientException(error.getError(),
buildError(error.getRequestId(), error.getMessage())));
@@ -1006,13 +1041,6 @@ public class ClientCnx extends PulsarHandler {
TimedCompletableFuture<LookupDataResult> future = new TimedCompletableFuture<>();
if (pendingLookupRequestSemaphore.tryAcquire()) {
- future.whenComplete((lookupDataResult, throwable) -> {
- if (throwable instanceof ConnectException
- || throwable instanceof PulsarClientException.LookupException
- || FutureUtil.unwrapCompletionException(throwable) instanceof TimeoutException) {
- pendingLookupRequestSemaphore.release();
- }
- });
addPendingLookupRequests(requestId, future);
ctx.writeAndFlush(request).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
@@ -1020,8 +1048,9 @@ public class ClientCnx extends PulsarHandler {
.attr("requestId", requestId)
.exceptionMessage(writeFuture.cause())
.log("Failed to send request to broker");
- getAndRemovePendingLookupRequest(requestId);
- future.completeExceptionally(writeFuture.cause());
+ if (removePendingRequest(requestId, future)) {
+ future.completeExceptionally(writeFuture.cause());
+ }
}
});
} else {
@@ -1069,8 +1098,8 @@ public class ClientCnx extends PulsarHandler {
.log("Received get topics of namespace success response from server");
CompletableFuture<GetTopicsResult> requestFuture =
- (CompletableFuture<GetTopicsResult>) pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ (CompletableFuture<GetTopicsResult>) pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.complete(new GetTopicsResult(topics,
success.hasTopicsHash() ? success.getTopicsHash() : null,
success.isFiltered(),
@@ -1089,8 +1118,8 @@ public class ClientCnx extends PulsarHandler {
long requestId = commandGetSchemaResponse.getRequestId();
CompletableFuture<CommandGetSchemaResponse> future =
- (CompletableFuture<CommandGetSchemaResponse>) pendingRequests.remove(requestId);
- if (future == null) {
+ (CompletableFuture<CommandGetSchemaResponse>) pendingRequests.get(requestId);
+ if (future == null || !removePendingRequest(requestId, future)) {
duplicatedResponseCounter.incrementAndGet();
log.warn().attr("requestId", requestId)
.log("Received unknown request id from server");
@@ -1104,10 +1133,10 @@ public class ClientCnx extends PulsarHandler {
checkArgument(state == State.Ready);
long requestId = commandGetOrCreateSchemaResponse.getRequestId();
CompletableFuture<CommandGetOrCreateSchemaResponse> future =
- (CompletableFuture<CommandGetOrCreateSchemaResponse>) pendingRequests.remove(requestId);
- if (future == null) {
+ (CompletableFuture<CommandGetOrCreateSchemaResponse>) pendingRequests.get(requestId);
+ if (future == null || !removePendingRequest(requestId, future)) {
duplicatedResponseCounter.incrementAndGet();
- log.warn().attr("requestId", requestId)
+ log.warn().attr("requestId", commandGetOrCreateSchemaResponse.getRequestId())
.log("Received unknown request id from server");
return;
}
@@ -1141,7 +1170,7 @@ public class ClientCnx extends PulsarHandler {
pendingRequests.put(requestId, future);
(flush ? ctx.writeAndFlush(requestMessage) : ctx.write(requestMessage)).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
- if (pendingRequests.remove(requestId, future) && !future.isDone()) {
+ if (removePendingRequest(requestId, future)) {
log.warn()
.attr("send", requestType.getDescription())
.exceptionMessage(writeFuture.cause())
@@ -1262,9 +1291,8 @@ public class ClientCnx extends PulsarHandler {
.attr("requestId", response.getRequestId())
.log("Received tc client connect response from server");
long requestId = response.getRequestId();
- CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);
-
- if (requestFuture != null && !requestFuture.isDone()) {
+ CompletableFuture<?> requestFuture = pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
if (!response.hasError()) {
requestFuture.complete(null);
} else {
@@ -1335,8 +1363,8 @@ public class ClientCnx extends PulsarHandler {
.log("Received watchTopicListSuccess response from server");
long requestId = commandWatchTopicListSuccess.getRequestId();
CompletableFuture<CommandWatchTopicListSuccess> requestFuture =
- (CompletableFuture<CommandWatchTopicListSuccess>) pendingRequests.remove(requestId);
- if (requestFuture != null) {
+ (CompletableFuture<CommandWatchTopicListSuccess>) pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.complete(new CommandWatchTopicListSuccess().copyFrom(commandWatchTopicListSuccess));
} else {
duplicatedResponseCounter.incrementAndGet();
@@ -1409,8 +1437,8 @@ public class ClientCnx extends PulsarHandler {
log.debug().attr("requestId", requestId).log("Received scalableTopicSubscribeResponse");
if (cmd.hasError()) {
- CompletableFuture<? extends Object> requestFuture = pendingRequests.remove(requestId);
- if (requestFuture != null && !requestFuture.isDone()) {
+ CompletableFuture<? extends Object> requestFuture = pendingRequests.get(requestId);
+ if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
requestFuture.completeExceptionally(new PulsarClientException(
"Scalable topic subscribe failed: " + cmd.getError()
+ (cmd.hasMessage() ? " - " + cmd.getMessage() : "")));
@@ -1421,8 +1449,9 @@ public class ClientCnx extends PulsarHandler {
@SuppressWarnings("unchecked")
TimedCompletableFuture<org.apache.pulsar.common.api.proto.ScalableConsumerAssignment>
requestFuture = (TimedCompletableFuture<
- org.apache.pulsar.common.api.proto.ScalableConsumerAssignment>) pendingRequests.remove(requestId);
- if (requestFuture == null || requestFuture.isDone()) {
+ org.apache.pulsar.common.api.proto.ScalableConsumerAssignment>) pendingRequests.get(requestId);
+ if (requestFuture == null || !removePendingRequest(requestId, requestFuture)) {
+ duplicatedResponseCounter.incrementAndGet();
log.warn().attr("requestId", requestId)
.log("Received scalable topic subscribe response for unknown / completed request");
return;
@@ -1794,10 +1823,9 @@ public class ClientCnx extends PulsarHandler {
continue;
}
TimedCompletableFuture<?> requestFuture = pendingRequests.get(request.requestId);
- if (requestFuture != null
- && !requestFuture.hasGotResponse()) {
- pendingRequests.remove(request.requestId, requestFuture);
- if (!requestFuture.isDone()) {
+ if (requestFuture != null && !requestFuture.hasGotResponse()) {
+ boolean removed = removePendingRequest(request.requestId, requestFuture);
+ if (removed) {
String timeoutMessage = request.requestType.getDescription() + " timeout";
if (requestFuture.completeExceptionally(new TimeoutException(timeoutMessage))) {
if (request.requestType == RequestType.Lookup) {
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 db1725089802..4413d614d1e2 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.CompletableFuture;
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;
@@ -46,6 +47,7 @@ import org.apache.pulsar.common.api.proto.CommandCloseConsumer;
import org.apache.pulsar.common.api.proto.CommandCloseProducer;
import org.apache.pulsar.common.api.proto.CommandConnected;
import org.apache.pulsar.common.api.proto.CommandError;
+import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.api.proto.ServerError;
@@ -373,6 +375,302 @@ public class ClientCnxTest {
});
}
+ /**
+ * 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"));
+ try {
+ 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);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
+ /**
+ * 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"));
+ try {
+ 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);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
+ /**
+ * 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"));
+ try {
+ 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);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
+ /**
+ * Test that when a failed lookup response is received,
+ * the semaphore permit is properly released via the unified cleanup primitive
+ * (removePendingRequest) without double-release.
+ */
+ @Test
+ public void testFailedLookupResponseReleasesSemaphore() throws Exception {
+ EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false,
+ new DefaultThreadFactory("testFailedLookupResponseReleasesSemaphore"));
+ try {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setOperationTimeoutMs(10_000);
+ conf.setKeepAliveIntervalSeconds(0);
+ ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
+ cnx.channelActive(ctx);
+ cnx.state = ClientCnx.State.Ready;
+
+ int initialPermits = cnx.getPendingLookupRequestSemaphore().availablePermits();
+
+ // Send a lookup request
+ CompletableFuture<BinaryProtoLookupService.LookupDataResult> future =
+ cnx.newLookup(null, 1L);
+
+ // Simulate a failed lookup response (CommandLookupTopicResponse with Failed status)
+ CommandLookupTopicResponse lookupResponse = new CommandLookupTopicResponse();
+ lookupResponse.setRequestId(1L);
+ lookupResponse.setResponse(CommandLookupTopicResponse.LookupType.Failed);
+ lookupResponse.setError(ServerError.ServiceNotReady);
+ lookupResponse.setMessage("Service not ready");
+ cnx.handleLookupResponse(lookupResponse);
+
+ // Verify the future completed exceptionally
+ try {
+ future.get(2, TimeUnit.SECONDS);
+ fail("Should have failed");
+ } catch (Exception e) {
+ // expected
+ }
+
+ // Verify semaphore is released back to initial permits (no leak, no double-release)
+ // Previously this would cause availablePermits to be initialPermits+1 due to double-release
+ Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), initialPermits);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
+ /**
+ * Test that when the connection is closed while an active lookup request is pending,
+ * the semaphore permit is properly released.
+ */
+ @Test
+ public void testActiveRequestConnectionCloseReleasesSemaphore() throws Exception {
+ EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false,
+ new DefaultThreadFactory("testActiveRequestConnectionCloseReleasesSemaphore"));
+ try {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setOperationTimeoutMs(10_000);
+ 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
+ CompletableFuture<BinaryProtoLookupService.LookupDataResult> future =
+ cnx.newLookup(null, 1L);
+
+ // Simulate connection close
+ cnx.channelInactive(ctx);
+
+ // Verify the future completed exceptionally with ConnectException
+ try {
+ future.get(2, TimeUnit.SECONDS);
+ fail("Should have failed");
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof PulsarClientException.ConnectException);
+ }
+
+ // Verify semaphore is released back to initial permits
+ Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), initialPermits);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
+ /**
+ * Test that when the connection is closed while a promoted waiting request is active,
+ * the semaphore permit is properly released and no leak occurs.
+ */
+ @Test
+ public void testPromotedWaitingRequestConnectionCloseReleasesSemaphore() throws Exception {
+ EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false,
+ new DefaultThreadFactory("testPromotedWaitingRequestConnectionCloseReleasesSemaphore"));
+ try {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setOperationTimeoutMs(10_000);
+ conf.setKeepAliveIntervalSeconds(0);
+ conf.setConcurrentLookupRequest(1); // Only allow 1 concurrent lookup
+ conf.setMaxLookupRequest(10);
+ 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 goes into the waiting queue
+ CompletableFuture<BinaryProtoLookupService.LookupDataResult> secondFuture =
+ cnx.newLookup(null, 2L);
+
+ // Simulate a failed lookup response for the first request, which promotes the second
+ CommandLookupTopicResponse failedResponse = new CommandLookupTopicResponse();
+ failedResponse.setRequestId(1L);
+ failedResponse.setResponse(CommandLookupTopicResponse.LookupType.Failed);
+ failedResponse.setError(ServerError.ServiceNotReady);
+ failedResponse.setMessage("Service not ready");
+ cnx.handleLookupResponse(failedResponse);
+
+ // Verify the first future completed exceptionally
+ assertTrue(firstFuture.isCompletedExceptionally());
+
+ // Wait for the second request to be promoted from the waiting queue
+ Awaitility.await().atMost(2, TimeUnit.SECONDS)
+ .until(() -> cnx.pendingRequests.containsKey(2L));
+
+ // Now close the connection while the promoted request is active
+ cnx.channelInactive(ctx);
+
+ // Verify the second future completed exceptionally
+ try {
+ secondFuture.get(2, TimeUnit.SECONDS);
+ fail("Second future should have failed");
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof PulsarClientException.ConnectException);
+ }
+
+ // Verify semaphore is released (1 permit for concurrentLookupRequest=1)
+ Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), 1);
+ });
+ } finally {
+ eventLoop.shutdownGracefully().sync();
+ }
+ }
+
private void withConnection(String testName, Consumer<ClientCnx> test) {
ThreadFactory threadFactory = new DefaultThreadFactory(testName);
EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);