[fix][client] Fix lookup permit double-release, waiting queue starvation and timeout-response races in ClientCnx

[geniusjoe]

main pr #25038

Motivation

PR #25038 fixed the semaphore leak issue by adding a TimeoutException check in the whenComplete callback of newLookup(). However, there are several remaining issues:

1. Waiting queue starvation on timeout: When a lookup request times out in checkRequestTimeout(), the code only calls pendingRequests.remove() and relies on the whenComplete callback to release the semaphore. While the semaphore is correctly released, getAndRemovePendingLookupRequest() — which is responsible for polling and dispatching the next waiting request — is never invoked on the timeout path. If all concurrent lookup slots time out simultaneously (e.g., during a broker GC pause), waiting requests remain stuck indefinitely.

2. Double-release of lookup semaphore: When a lookup receives a failed response, getAndRemovePendingLookupRequest() releases the permit (or transfers it to the waiting queue). Then handleLookupResponse() completes the future with LookupException, which triggers the whenComplete callback registered in ClientCnx.newLookup() to call pendingLookupRequestSemaphore.release() again — because LookupException is one of the exception types that the callback explicitly releases for. This causes availablePermits() to exceed its configured initial value.

3. Unclear permit ownership for promoted waiting requests: When a request is promoted from the waiting queue, its permit ownership is ambiguous. If the promoted request's write fails or it times out, it's unclear which path is responsible for releasing the permit.

4. Race conditions between timeout and response paths: The timeout handler and response handler could concurrently process the same request, leading to spurious duplicatedResponseCounter increments and potential double-completion.

Modifications

Let the permit ownership follow the active entry in pendingRequests:

1. Only the path that successfully removes the entry from pendingRequests owns the cleanup responsibility. Introduced removePendingRequest(requestId, expectedFuture) which uses pendingRequests.remove(requestId, expectedFuture) as the single CAS contention point. Only the thread that wins the CAS can complete the future. For lookup requests, the winner additionally releases or transfers the permit.

2. All pending request completion paths go through the same removePendingRequest primitive. Timeout, lookup response, write failure, CommandError, channel close, and all other response handlers (ack response, producer success, get schema, get topics, etc.) now call removePendingRequest uniformly. Only the CAS winner can complete the future. For lookup requests specifically, the winner additionally calls releasePermitAndDriveWaitingQueue() to either transfer the permit to the next waiting request or release it back to pendingLookupRequestSemaphore.

3. The helper uses remove(requestId, expectedFuture) and returns whether ownership was successfully obtained. Added pendingLookupRequestIds (a ConcurrentHashMap-backed Set) to track which requestIds belong to lookup requests, so that removePendingRequest can correctly identify and release lookup semaphore permits.

4. No longer guess whether the permit needs to be released based on the type of future exception. Removed the future.whenComplete callback in newLookup() that previously inferred permit ownership from exception type (ConnectException / LookupException). Permit release is now exclusively handled inside removePendingRequest.

5. Added permit invariant tests for failed lookup response, active request connection close, and promoted waiting request connection close. New tests: testFailedLookupResponseReleasesSemaphore, testActiveRequestConnectionCloseReleasesSemaphore, testPromotedWaitingRequestConnectionCloseReleasesSemaphore.

This resolves timeout starvation, double-release, and the unclear permit ownership issue for promoted waiting requests on certain exception paths at once.

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.
• testFailedLookupResponseReleasesSemaphore — verifies that a failed CommandLookupTopicResponse correctly releases exactly one permit without over-release.
• testActiveRequestConnectionCloseReleasesSemaphore — verifies that connection close correctly releases the permit for an active lookup request.
• testPromotedWaitingRequestConnectionCloseReleasesSemaphore — verifies that connection close correctly releases the permit for a promoted waiting request.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Local workflow:

geniusjoe#1

[geniusjoe]

@congbobo184
Hi! Since you're familiar with this area from #25038, would you mind taking a look at this PR when you have a chance? It optimizes the timeout path to also drive the waiting lookup queue. Thanks in advance

[Denovo1998]

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.

[geniusjoe]

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.

@Denovo1998
Thanks for the detailed review! All points have been addressed in the latest push.

The core change introduces a unified removePendingRequest(requestId, expectedFuture) method
that uses pendingRequests.remove(requestId, expectedFuture) as the single CAS contention
point. Only the winner owns cleanup responsibility including permit release. Timeout, lookup
response, write failure, CommandError, and channel close all go through this same primitive now.

One implementation note: since pendingRequests stores all request types in a single map
without preserving the command type, I added a pendingLookupRequestIds set
(ConcurrentHashMap-backed) to track which requestIds belong to lookup requests. Inside
removePendingRequest, after a successful CAS removal, it checks whether the requestId
is in this set — if so, it releases the semaphore permit and drives the waiting queue;
otherwise it simply returns without any permit-related logic.

For consistency, all response handlers that previously called pendingRequests.remove(requestId)
directly have been migrated to use removePendingRequest as the single entry point — this also
prevents concurrent paths (e.g. timeout vs. late response) from completing the same future twice.

The whenComplete callback that inferred permit ownership from exception type has been
removed entirely. All 3 requested permit invariant tests have been added. Could you take
another look and let me know if this approach aligns with your expectations?

[geniusjoe]

@Denovo1998 PTAL, both comments have been addressed.

[Denovo1998]

LGTM. Update the pr description.

[geniusjoe]

LGTM. Update the pr description.

@Denovo1998
Thanks for the review! PR description has been updated to reflect the current implementation.

[hanmz]

lgtm