Denovo1998 commented on code in PR #26143:
URL: https://github.com/apache/pulsar/pull/26143#discussion_r3565513829
##########
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();
Review Comment:
getAndRemovePendingLookupRequest() has already released or transferred the
active lookup permit before handleLookupResponse() completes a failed response
with LookupException. This callback then releases the same permit again. A
failed CommandLookupTopicResponse increases availablePermits() from 5000 to
5001 in a minimal test. Could we centralize permit release/transfer in the
successful pending-map removal path instead of inferring ownership from the
exception type?
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java:
##########
@@ -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();
Review Comment:
Please wrap the event loop lifetime in try/finally so it is also shut down
when an assertion or timed get() fails. It would also be safer to wait for
graceful shutdown to complete; otherwise these timeout tests can leave threads
behind and contaminate later tests.
Those below are also needed.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java:
##########
@@ -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))) {
Review Comment:
Could this path use an expected-future removal and only complete the timeout
when it actually wins the removal race? At present we get() the future, call an
unconditional remove-by-id helper, ignore its result, and may still complete
the old reference with a timeout after another path removed it. The helper also
increments duplicatedResponseCounter when this race is lost, although no
duplicate response was received.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
