[joviegas]

Motivation and Context

The SQS receive batch manager creates one receive buffer per queue URL and keeps it until the client closes. The number of buffers has no limit, and closing the client does not return prefetched messages to the queue until their visibility timeout expires.

This change caps the number of receive buffers, evicts idle buffers at the cap, and returns prefetched messages to the queue on close.

Modifications

• Buffer cap: New ReceiveBufferMap holds at most 10,000 buffers (RequestBatchConfiguration.DEFAULT_MAX_BATCH_KEYS). At the cap, a new queue evicts the buffer idle the longest. If every buffer is in use, the request fails with IllegalStateException.
• Buffer creation: ReceiveMessageBatchManager creates a buffer only after the queue's attributes load. A queue URL that fails to load gets no buffer. Concurrent first requests for a queue share one attribute load.
• Buffer key: Buffers are keyed by queue URL. Previously the key also included the request override configuration hash.
• Close: After close, new requests fail with IllegalStateException. ReceiveQueueBuffer fails requests that arrive during close and returns messages from receives in flight at close to the queue.
• Visibility reset: ReceiveSqsMessageHelper.clear() drains messages with poll(), so a concurrent removeMessage() and clear() handle each message once. It logs a warning when ChangeMessageVisibilityBatch reports failed entries or throws synchronously.

Testing

• ReceiveBufferEvictionTest (real buffers): 20,000 distinct queues hold at most 10,000 buffers; eviction returns held messages; missing queues take no slot; a buffer evicted mid-receive returns its messages.
• ReceiveMessageBatchManagerLimitTest: eviction order, every buffer in use, attribute load failure and retry, shared attribute loads, close racing new requests, and concurrent use never exceeding the cap.
• ReceiveQueueBufferTest: requests after close fail; receives in flight at close return their messages.
• ReceiveSqsMessageHelperTest: warnings for failed and thrown visibility resets; clear() racing removeMessage() over 500 rounds.

SqsAsyncBatchManager.receiveMessage performance

Scenario	Before (ops/s)	After (ops/s)	Delta
16 threads, 1 queue, 0 ms	79,722	78,014	-2.1%
16 threads, 1 queue, 2 ms	4,782	4,776	-0.1%
16 threads, 1 queue, 20 ms	501	502	+0.3%
16 threads, 1 queue, 50 ms	372	370	-0.4%
16 threads, 16 queues, 0 ms	137,666	133,431	-3.1%
16 threads, 16 queues, 2 ms	10,067	10,052	-0.1%
16 threads, 16 queues, 20 ms	1,134	1,148	+1.2%
16 threads, 16 queues, 50 ms	523	518	-1.1%
1 thread, 1 queue, 0 ms	28,327	27,700	-2.2%
1 thread, 1 queue, 2 ms	480	477	-0.6%
1 thread, 1 queue, 20 ms	49.8	51.3	+3.0%
1 thread, 1 queue, 50 ms	36.4	34.5	-5.1%
1 thread, 16 queues, 0 ms	28,291	27,939	-1.2%
1 thread, 16 queues, 2 ms	477	475	-0.2%
1 thread, 16 queues, 20 ms	49.4	49.5	+0.1%
1 thread, 16 queues, 50 ms	29.5	32.0	+8.4%

Before: master 220e5a84340. After: fix branch with latest changes. JMH on a 16-core dev desk; ms is the stub SQS response delay.

License

• I confirm that this pull request can be released under the Apache 2 license

[davidh44]

A waiter on a shared lookup can add a manager that eviction already closed. Waiters run add(queueUrl, manager) one at a time when the lookup completes. If an earlier waiter's request finishes and its entry is evicted at the cap before a later waiter runs, that later add inserts a new entry wrapping the closed manager (ReceiveBufferMap.java:81). After that every receive for the queue fails with "The client has been shut down." until the entry happens to be evicted again. Could add reject a closed manager, so the queue gets reloaded on the next request instead of staying broken?

[joviegas]

Fixed. Rejecting a closed manager inside add still leaves a race, because eviction removes the entry under the lock but closes the manager after releasing it. Instead, the shared lookup now completes with the Entry, eviction marks it evicted under the lock, and add returns null for an evicted candidate. The request then drops the stale lookup and loads the queue again. I added batchRequest_whenSharedLookupWaiterRunsAfterEviction_loadsNewBuffer, which failed before the fix.

[davidh44]

When all 10,000 buffers are in use the receive fails, but sqsClient.receiveMessage(request) could serve it unbuffered, the same way ineligible requests already go. That would make hitting the limit a performance hit instead of an error for the caller. What is the reasoning behind failing instead?

[joviegas]

A request fails only when 10,000 different queues each have a receive call waiting at the same moment. Queues with no waiting call are evicted to make room. This matches the send-side batch managers, which fail at the same 10,000 limit (BatchingMap throws IllegalStateException). A silent fallback would turn every receive into a separate SQS call without telling the user that batching stopped. Failing makes the limit visible.

[davidh44]

Should we callout the behavior changes? 1)receiveMessage on a queue that doesn't exist now fails with QueueDoesNotExistException even when waitTimeSeconds is set. Before, it returned empty responses. 2) After close() every receiveMessage fails, including unbuffered ones that used to go straight to SQS.

[joviegas]

Updated the changelog entry to call out both behavior changes.

[davidh44]

A few of the new tests cover the same paths as other tests, should we trim them?

1. This one only has a single idle entry, so it reduces to "evict the only idle buffer". batchRequest_whenIdleBufferIsReused_isNotEvictedWhileInUse covers the harder case where the oldest idle buffer goes back into use and must be skipped, and batchRequest_whenEveryBufferIsInUse_failsWithoutCallingSqs already shows busy buffers are never evicted.
2. ReceiveBufferEvictionTest.batchRequest_whenBufferEvictedWithReceiveInFlight_returnsItsMessagesWhenItCompletes tests the same reportBatchFinished shutdown branch as ReceiveQueueBufferTest.close_whenReceiveInFlight_returnsItsMessagesWhenItCompletes. The only extra it shows is that eviction calls close(), which batchRequest_whenAtLimit_evictsOldestIdleBuffersAndReturnsTheirHeldMessages already covers. I'd keep the buffer-level one since it doesn't need polling.
3. ReceiveSqsMessageHelperTest.clear_whenChangeVisibilityFutureFails_logsWarn asserts the same warning on the same failed-future branch as the existing ReceiveQueueBufferTest.visibilityTimeOutErrorsAreLogged.

[joviegas]

Removed all three, and kept ReceiveQueueBufferTest.close_whenReceiveInFlight_returnsItsMessagesWhenItCompletes for the in-flight close path.