[fix][client] Fix waiting lookup requests not dispatched on timeout in ClientCnx#26143
[fix][client] Fix waiting lookup requests not dispatched on timeout in ClientCnx#26143geniusjoe wants to merge 1 commit into
Conversation
|
@congbobo184 |
Denovo1998
left a comment
Denovo1998
left a comment
There was a problem hiding this comment.
I think it's best to let the permit ownership follow the active entry in pendingRequests:
-
Only the path that successfully removes the active lookup from pendingRequests has the right to release or transfer the permit.
-
Timeout, lookup response, write failure, CommandError, and channel close all go through the same lookup cleanup primitive.
-
The helper uses remove(requestId, expectedFuture) and returns whether ownership was successfully obtained.
-
No longer guess whether the permit needs to be released based on the type of future exception.
-
Add permit invariant tests for failed lookup response, active request connection close, and promoted waiting request connection close.
This might resolve timeout starvation, double-release, and the unclear permit ownership issue for promoted waiting requests on certain exception paths at once.
| future.whenComplete((lookupDataResult, throwable) -> { | ||
| if (throwable instanceof ConnectException | ||
| || throwable instanceof PulsarClientException.LookupException | ||
| || FutureUtil.unwrapCompletionException(throwable) instanceof TimeoutException) { | ||
| || throwable instanceof PulsarClientException.LookupException) { | ||
| pendingLookupRequestSemaphore.release(); |
There was a problem hiding this 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?
| 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))) { |
There was a problem hiding this 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.
| assertEquals(cnx.getPendingLookupRequestSemaphore().availablePermits(), initialPermits); | ||
| }); | ||
|
|
||
| eventLoop.shutdownGracefully(); |
There was a problem hiding this 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.
2 participants
main pr #25038
Motivation
PR #25038 fixed the semaphore leak issue by adding a
TimeoutExceptioncheck in thewhenCompletecallback ofnewLookup(). However, it did not fully address the waiting queue starvation problem.When a lookup request times out in
checkRequestTimeout(), the code only callspendingRequests.remove()and relies on thewhenCompletecallback to release the semaphore. While the semaphore is correctly released,getAndRemovePendingLookupRequest()— which is responsible for polling and dispatching the next waiting request fromwaitingLookupRequests— is never invoked on the timeout path.This means: if all concurrent lookup slots time out simultaneously (e.g., during a broker GC pause or network partition), the waiting requests remain stuck indefinitely even though semaphore permits become available. They will never be dispatched until a new non-timeout lookup response arrives.
Modifications
Optimized the timeout handling path in
ClientCnx.java:checkRequestTimeout(): ForLookup-type requests, usegetAndRemovePendingLookupRequest()instead of plainpendingRequests.remove(). This ensures the next waiting request inwaitingLookupRequestsis polled and dispatched upon timeout, rather than just releasing the semaphore.newLookup()whenCompletecallback: Removed theTimeoutExceptioncheck sincegetAndRemovePendingLookupRequest()already handles semaphore release and queue driving. Keeping it would cause a double-release.Verifying this change
This change added tests and can be verified as follows:
testLookupTimeoutReleasesSemaphore— verifies that the semaphore is properly released after a lookup timeout.testLookupTimeoutDrivesWaitingQueue— verifies that when a concurrent lookup times out, the next waiting request in the queue is dispatched.testMultipleLookupTimeoutsNoSemaphoreLeak— verifies that with multiple concurrent and queued lookups all timing out, every request eventually completes and no semaphore permits are leaked.Does this pull request potentially affect one of the following parts:
Local workflow:
geniusjoe#1