Receive Batch Manager Implementation #5488

joviegas · 2024-08-08T19:53:48Z

Motivation and Context

This change is required to efficiently manage the receiving of message batches from AWS SQS, ensuring proper handling of message visibility timeout and error scenarios. It solves the problem of handling asynchronous message retrieval and processing, managing visibility timeouts, and handling exceptions during the process.

Modifications

AsyncReceiveMessageBatch Class:
- Forms a ReceiveMessageRequest based on configuration settings.
- Collects messages from an AWS SQS queue.
- Handles exceptions during the process.
- Manages message visibility timeout by tracking the visibility deadline and expiring messages if not processed in time.
- Supports clearing messages in the batch and changing their visibility as needed.
ReceiveMessageCompletableFuture Class:
- Manages the completion of the receive message future with success or failure.
- Starts a wait timer for the receive message request and handles timeout scenarios.
ReceiveQueueBuffer Class:
- Manages the queue buffer for received messages.
- Handles the reception of messages and the spawning of additional receive tasks.
- Manages the clearing of finished tasks and handling of futures for received messages.
- Supports shutdown functionality, ensuring all futures are completed exceptionally upon shutdown.
ResponseBatchConfiguration Class:
- Configures various batch settings such as visibility timeout, long poll settings, and adaptive prefetching.
QueueAttributesManager Class:
- Manages SQS queue attributes.
- Fetches and caches attributes like visibility timeout and receive message wait time.
- Ensures only a single call is made to fetch the attributes, with proper handling of concurrent requests and failures.

Design Details

Asynchronous Handling:
- The AsyncReceiveMessageBatch uses asynchronous handling for receiving messages from SQS. It leverages CompletableFuture to manage asynchronous operations and handle results or exceptions.
Visibility Timeout Management:
- The class tracks the visibility deadline for messages to ensure they are reprocessed if not handled within the specified time. This is managed using visibilityDeadlineNano.
Error Handling:
- The AsyncReceiveMessageBatch and QueueAttributesManager classes handle exceptions by setting the future to complete exceptionally and resetting state where necessary.
Concurrent Task Management:
- ReceiveQueueBuffer uses a concurrent queue to manage finished tasks and inflight receive message batches, ensuring thread-safe operations.
- QueueAttributesManager uses an AtomicReference to manage the cached future for fetching queue attributes, ensuring that only one request is made even under concurrent access.
Adaptive Prefetching:
- The ResponseBatchConfiguration supports adaptive prefetching, adjusting the number of batches based on the load and configuration.

Testing

Unit tests were added for AsyncReceiveMessageBatch to cover:
- Message receiving and handling.
- Visibility timeout management.
- Error handling during message receiving.
Tests for ReceiveQueueBuffer include:
- Handling multiple receive message requests.
- Managing finished tasks and inflight batches.
- Testing shutdown behavior.
QueueAttributesManager tests:
- Fetching and caching queue attributes.
- Handling exceptions during attribute fetching.
- Ensuring concurrent requests are handled correctly and only a single call is made to fetch attributes.

License

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

…_buffer_queue

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

dagnir · 2024-08-08T22:48:43Z

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

+                                      messages = new ArrayList<>(response.messages());
+                                  }
+                                  open.set(true);
+                                  return this;
+                              });
+        } finally {
+            visibilityDeadlineNano = System.nanoTime() + visibilityTimeout.toNanos();
+        }


What happens if this gets called multiple times? Seems like these can get out of sync; for example one request might succeed so messages has content, but another one fails so exception is also non-null.

This method is called per instance of AsyncReceiveMessageBatch. Each thread creates its own instance, so there’s no case where it can go out of sync. Since this internal API is called per instance, I can’t think of a scenario where a single instance is created and accessed by multiple threads. If you see any such case, please let me know

dagnir · 2024-08-08T22:49:07Z

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

+        this.config = config;
+    }
+
+    public CompletableFuture<AsyncReceiveMessageBatch> asyncReceiveMessage() {


Do we support calling this multiple times?

Every spawn calls create as new instances of AsyncReceiveMessageBatch

AsyncReceiveMessageBatch asyncReceiveMessageBatch = new AsyncReceiveMessageBatch( queueUrl, sqsClient, visibilityTimeoutNanos, config); inflightReceiveMessageBatches.incrementAndGet(); asyncReceiveMessageBatch.asyncReceiveMessage() .whenComplete((response, exception) -> reportBatchFinished(response));

So two simultaneous calls will create two instances and do two separate reportBatchFinished which is expected.
However AsyncReceiveMessageBatch is handled for Thread safety

By making

private volatile Queue<Message> messages = new ConcurrentLinkedQueue<>(); // Thread-safe queue

We are now able to run this mutiple times in a thread safe way

dagnir · 2024-08-08T22:52:55Z

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

+
+    public AsyncReceiveMessageBatch(ScheduledExecutorService scheduledExecutorService, String queueUrl,
+                                    SqsAsyncClient asyncClient, Duration visibilityTimeout, ResponseBatchConfiguration config) {
+        this.scheduledExecutorService = scheduledExecutorService;


This looks unused

dagnir · 2024-08-08T22:57:46Z

...test/java/software/amazon/awssdk/services/sqs/BatchManager/AsyncReceiveMessageBatchTest.java

+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.sqs.BatchManager;


capitalized package name

dagnir · 2024-08-08T22:58:59Z

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

+    private final ResponseBatchConfiguration config;
+    private final AtomicBoolean open = new AtomicBoolean(false);
+    private volatile Throwable exception;
+    private List<Message> messages;


At the very least, this probably needs to be volatile too

Okay, made it volatile and used CopyOnWriteArrayList

dagnir · 2024-08-08T23:00:02Z

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

+        return asyncClient.changeMessageVisibilityBatch(batchRequest);
+    }
+
+    private void checkIfOpen() {


I'm not really sure what the purpose of ensure this is "open" should be. What does "open" indicate?

removed this , this was redundant check to make sure we donot access the exception or messages while the receieveMessage call has not happened or in progress.

dagnir · 2024-08-08T23:01:35Z

Still looking, giving some early feedback

...n/java/software/amazon/awssdk/services/sqs/internal/batchmanager/QueueAttributesManager.java

dagnir · 2024-08-08T23:22:01Z

...n/java/software/amazon/awssdk/services/sqs/internal/batchmanager/QueueAttributesManager.java

+        });
+
+        resultFuture.whenComplete((r, t) -> {
+            if (resultFuture.isCancelled()) {


minor: probably better to just check instanceof on t. If this future was cancelled, t will have type CancellationException.

zoewangg

Still going through the PR, left some initial comments

...c/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

...java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

...n/java/software/amazon/awssdk/services/sqs/internal/batchmanager/QueueAttributesManager.java

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

zoewangg

Left some comments, still not finished review

...c/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

zoewangg · 2024-08-16T17:29:15Z

...c/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

+        }
+
+        /**
+         * Define the message attributes receive calls will request. Only receive message requests that request the same set of


I understand this is what used in v1, but should we use something like .receiveMessageRequestTransformer to make it easier for customers to change other things in the request? And we can get rid of the other settings like maxBatchItems?

Prior art: https://github.com/aws/aws-sdk-java-v2/blob/master/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/model/DownloadDirectoryRequest.java#L329

Thanks, Zoe. If that’s the case, we’ll need to change the behavior of the feature. Specifically, when fetching SQS messages, we would fetch all the attributes and then filter them at the request level based on the attributes passed in the request. This change would have two main impacts:

1. Diverging from V1 behavior. 2. Requiring the retrieval of all attributes, even when the user has configured only a specific number of attribute names in the configuration.

It’s important to note that we currently only respect the maxNumberOfMessages and waitTimeSeconds parameters in the request, which are specific to the request. The other parameters are referenced from the configuration settings and are queue-level settings.

Given this, we have the following options:

• We could remove receiveAttributeNames for now and add it later if there is a user request. • We could keep it the same as V1 and provide a configuration option, as it currently exists. • If we provide a request-level option like receiveMessageRequestTransformer, we could pull out the entire attributes map from SQS and then filter it while completing the future.

From my observation of the attributeMap, I think the current behavior—where configuration-level attributeNames are used—is intended to optimize the receiveMessage payload when it comes to the map.

Also, in my opinion, if batchManager is something I expect to operate at the batch level, then if users want specific attributes, they can use the normal receiveMessage API. This way, we can avoid complications in logic where we complete the futures and keep it as simple as V1.

Will make sure will discuss in Surface API review meeting

zoewangg · 2024-08-16T17:38:43Z

...sqs/src/main/java/software/amazon/awssdk/services/sqs/batchmanager/SqsAsyncBatchManager.java

@@ -85,6 +86,29 @@ default CompletableFuture<ChangeMessageVisibilityResponse> changeMessageVisibili
        throw new UnsupportedOperationException();
    }

+
+    default CompletableFuture<SendMessageResponse> sendMessage(Consumer<SendMessageRequest.Builder> sendMessageRequest) {


Missing javadocs

my bad I raised a separate PR for this and accidently pushed some changes

zoewangg · 2024-08-16T21:52:52Z

...n/java/software/amazon/awssdk/services/sqs/internal/batchmanager/QueueAttributesManager.java

+                        queueAttributeMap.set(null); // Reset on failure
+                        newFuture.completeExceptionally(t); // Complete the future exceptionally
+                    } else {
+                        newFuture.complete(r); // Complete the future with the result


Can we remove the comments? The code seems to be self explanatory to me and having comments here clutter the code a bit IMO.

zoewangg · 2024-08-16T22:54:35Z

...n/java/software/amazon/awssdk/services/sqs/internal/batchmanager/QueueAttributesManager.java

+        future.whenComplete((r, t) -> {
+            if (t != null) {
+                queueAttributeMap.set(null); // Reset the future on failure
+            }
+        });


Do we need this here? It seems we nullify it on line 119

good catch , its not required.

zoewangg · 2024-08-16T23:15:04Z

...main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveBatchManager.java

+                                     });
+    }
+
+    public void shutdown() {


Can we follow v2 style - make it extend SdkAutoCloseable and implement close?

...ftware/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveMessageCompletableFuture.java

zoewangg · 2024-08-16T23:24:53Z

...main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveBatchManager.java

+                                         ReceiveMessageCompletableFuture receiveMessageFuture =
+                                             new ReceiveMessageCompletableFuture(numMessages, waitTimeMs);
+                                         receiveQueueBuffer.receiveMessage(receiveMessageFuture);
+                                         receiveMessageFuture.startWaitTimer(executor);


Instead of writing timer logic, can we use CompletableFuture#applyToEither or CompletebleFuture#anyOf instead where one is waiting for the response, the other is completing the future with empty response after the timeout.

We may be able to get rid of ReceiveMessageCompletableFuture class if the purpose of it is to encapsulate timer logic.

…_buffer_queue

zoewangg · 2024-08-17T00:14:40Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+    }
+
+    private void satisfyFuturesFromBuffer() {
+        pruneExpiredTasks();


It seems we can check expired tasks where we are processing future, i.e., line 170?

zoewangg · 2024-08-20T20:17:28Z

...c/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

+    /**
+     * @return the optional maximum amount of time that an outgoing call waits to be batched with messages of the same type.
+     */
+    public Duration maxBatchOpenDuration() {


We don't need to change this as part of this PR; I'm just writing down alternatives batchOpenTimeout

Will change this in next PR where I will handle Batch Size related comments from Surface area review

zoewangg · 2024-08-20T20:29:38Z

...c/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

+     * @return the message attributes receive calls will request.
+     */
+    public List<String> receiveMessageAttributeNames() {
+        return receiveMessageAttributeNames == null ? Collections.emptyList() :


Collections.unmodifiableList(Collections.emptyList())? Can we do the wrap in the ctor so that we don't need to wrap every time getter is invoked?

Handled this is builders() now

zoewangg · 2024-08-22T16:23:26Z

...main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveBatchManager.java

+        if (receiveQueueBuffer.isShutDown()) {
+            throw new IllegalStateException("The client has been shut down.");
+        }
+        int numMessages = Optional.ofNullable(rq.maxNumberOfMessages()).orElse(10);


nit: can we define all magic numbers in a single class?

Added SQSMessageDefaults

zoewangg · 2024-08-22T19:58:47Z

...va/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveMessageBatchManager.java

+    }
+
+    public CompletableFuture<ReceiveMessageResponse> batchRequest(ReceiveMessageRequest request) {
+        return canBeRetrievedFromQueueBuffer(request)


Can we add some debug logs to help troubleshoot in the future, i.e., when it decides to retrieve from the buffer or send it through the SDK client? Example: https://github.com/aws/aws-sdk-java-v2/blob/master/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/UploadWithKnownContentLengthHelper.java#L75

I think its safe to add only in canBeRetrievedFromQueueBuffer false case , else we will get lot of logs , added them now

zoewangg · 2024-08-22T20:20:49Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+        }
+
+        int desiredBatches = determineDesiredBatches();
+        if (finishedTasks.size() >= desiredBatches) {


ConcurrentLinkedQueue#size() is expensive, we should probably have a separate counter. https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentLinkedQueue.html#size--

Maximum number of Task in this case can be 10 since we decided to set the internal default to 10.
Adding separate counter will require to add synchronous blocks to that we atomically add and increment .
Thus considering the o(n) which will be max 10 I donot think we would require separate counters

zoewangg · 2024-08-22T20:22:34Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+            return;
+        }
+
+        queueAttributesManager.getVisibilityTimeout().thenAcceptAsync(visibilityTimeout -> {


Is there any reason we want to offload this to another executor? Offloading it to a thread may add more latency.

No reason , modified it

zoewangg · 2024-08-22T20:23:19Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+        queueAttributesManager.getVisibilityTimeout().thenAcceptAsync(visibilityTimeout -> {
+            int max = Math.max(config.maxInflightReceiveBatches(), 1);
+            int toSpawn = max - inflightReceiveMessageBatches.get();
+            if (toSpawn > 0) {


Should we spawn toSpawn worth of messages instead of just 1?

Nope , we should spawn exactly 1 , we donot want to spawn N number of messages but we want to minimize the number of spawned messages, that decision is taken based on inflight and unprocessed batches

Just for my own understanding, could you elaborate on why we don't want to spawn N number of messages?

The main purpose of the Automatic Batch Manager is to reduce the number of Service API calls.

We initiate an additional call at the end of an existing call or when a response is received to pre-fill our internal buffers. This is necessary because SQS doesn’t always have messages immediately available; it takes some time for the servers to populate the messages.

In most cases, users perform a Receive call with the default message size of 1. However, in the background, we fetch the maximum number of messages (10) in advance. This means we don't need N individual calls, but rather just one additional call to pre-fetch those messages. If we were to make N calls, we could end up with a maximum of N*10 messages.

Gotcha, thanks for the clarification

zoewangg · 2024-08-22T20:53:30Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+                }
+            });
+        } finally {
+            processingFutures.set(false);


We should probably double check again there's no task that needs to be published before releasing the flag.

private void satisfyFuturesFromBuffer() { do { if (!processingFutures.compareAndSet(false, true)) { return; } try { FutureRequestWrapper future = futures.poll(); if (!future.future.isDone()) { fulfillFuture(future); } } finally { processingFutures.set(false); } } while (shouldProcessMore()); }

zoewangg · 2024-08-22T20:54:18Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+        return desiredBatches;
+    }
+
+    private void fulfillFuture(FutureRequestWrapper futureWrapper, ReceiveSqsMessageHelper messageHelper) {


Suggesting calling finishedTasks.peek() directly instead of passing messageHelper in this method since it already has logic to use it.

private void fulfillFuture(FutureRequestWrapper futureWrapper)

zoewangg · 2024-08-22T23:25:51Z

...s/sqs/src/test/java/software/amazon/awssdk/services/sqs/batchmanager/SampleBatchManager.java

@@ -13,13 +13,12 @@
 * permissions and limitations under the License.
 */

-package software.amazon.awssdk.services.sqs.BatchManager;
+package software.amazon.awssdk.services.sqs.batchmanager;


Do we still need this class?

To test abstract class RequestBatchManager we need this class

zoewangg · 2024-08-23T17:23:51Z

.../java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveSqsMessageHelper.java

+                        // Logging an error is sufficient here as this is an asynchronous cleanup activity,
+                        // and there are no dependent tasks waiting for its completion.
+                        if (t != null) {
+                            log.error(() -> "Could not change visibility for queue " + queueUrl, t);


Should this be warn instead? Do customers need to retry this by themselves? Can we add more instructions here? Example: https://github.com/aws/aws-sdk-java-v2/blob/master/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/GenericMultipartHelper.java#L121

Yeap, made it to warn.
No customer need not retry , the SQS will make those messages visible after visibilty timeout is over

zoewangg · 2024-08-23T17:24:42Z

.../java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveSqsMessageHelper.java

+
+    public void clear() {
+        if (!isEmpty()) {
+            Optional.ofNullable(nackMessages())


nit: can we just use the good ol' null check instead of wrapping it with Optional.ofNullable (extra object)?

zoewangg · 2024-08-23T17:25:16Z

.../java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveSqsMessageHelper.java

+    public void clear() {
+        if (!isEmpty()) {
+            Optional.ofNullable(nackMessages())
+                    .ifPresent(future -> future.whenComplete((r, t) -> {


nit: if we are only handling exception case, we can just use exceptionally

.exceptionally(throwable -> { }

zoewangg · 2024-08-23T17:26:48Z

.../java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveSqsMessageHelper.java

+        return asyncClient.changeMessageVisibilityBatch(batchRequest);
+    }
+
+    public Integer messagesSize() {


Do we need this method? messages.size() is expensive

This message is very helpful for testing, this is internal API and used only for testing , I added @SdkTestInternalApi just to convey its only for testing

zoewangg · 2024-08-23T17:28:01Z

.../java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveSqsMessageHelper.java

+        return exception;
+    }
+
+    public void setException(Throwable exception) {


private? or maybe just remove this method and call this.exception = exception directly

sonarqubecloud · 2024-08-27T00:50:42Z

Quality Gate failed

Failed conditions
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarCloud

Catch issues before they fail your Quality Gate with our IDE extension SonarLint

joviegas · 2024-08-27T01:30:47Z

@zoewangg merge for now since I had to raise Surface API related PR , please approve this or let me know if more comments

zoewangg · 2024-09-03T19:50:27Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+        }
+
+        public ReceiveQueueBuffer build() {
+            return new ReceiveQueueBuffer(executor, sqsClient, config, queueUrl, queueAttributesManager);


Should we do return new ReceiveQueueBuffer(this);?

zoewangg · 2024-09-03T19:54:32Z

.../main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ReceiveQueueBuffer.java

+        queueAttributesManager.getVisibilityTimeout().thenAcceptAsync(visibilityTimeout -> {
+            int max = Math.max(config.maxInflightReceiveBatches(), 1);
+            int toSpawn = max - inflightReceiveMessageBatches.get();
+            if (toSpawn > 0) {


Just for my own understanding, could you elaborate on why we don't want to spawn N number of messages?

zoewangg · 2024-09-03T19:57:05Z

...c/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/SqsMessageDefault.java

+@SdkInternalApi
+public final class SqsMessageDefault {
+
+    public static final int MAX_SUPPORTED_SQS_RECEIVE_MSG = 10;


There's also ResponseBatchConfiguration, should we add this constant to that class?

joviegas · 2024-09-03T20:59:53Z

Raised #5550 for above comments

zoewangg

LGTM with the latest change

* Codegenerate BatchManager API under AsyncClient and Initail Interfaces for BatchManager (#5321) * Codegenerate BatchManager API under AsyncClient and Add empty initial Batchmanager interfaces and Implementations * Addressed review comments * Added Internal classes required for BatchManager Implementation * Revert "Added Internal classes required for BatchManager Implementation" This reverts commit 318969b. * Internal classes and RequestBatchManager Impelementation (#5418) * Added Internal classes required for BatchManager Implementation * Added Batch Send Implementation * Handled review comments * Handled review comments * Handled review comments * Made RequestBatchManager class a Abstract class * Checkstyle issues * Removed unused methods * New lines removed * Made public static to private state for sqsBatch functions * Constants added * Sonar cloud issues fixed * commit to check why test on codebuild * Increased Timeouts for get * Added abstract methods * Handled comments to remove Builders * Handled comments to take care when batchmanager closed while pending requests * Handled comments * Checkstyle issue * Added Consumer builders args for existing APIs of BatchManager (#5514) * Receive Batch Manager Implementation (#5488) * Add Recieve Buffer Queue And its related configuration * Update ReceiveBatch manager * Recieve Batch Manager Implementation * Receive Batch Manager Implemetation * Handled review comments * Checkstyle failure * Flsky test case fixed * Flaky test case fixed * Hamdled review comments * Handled comments * Removed ReceiveMessageCompletableFuture * SdkClosable implemented * Added ReceiveMessageBatchManager class for completeness * Checkstyle issues * Null checks * Handled comments from Zoe * Updated the defaults to 50ms same as V1 after surface area review * Revert "Updated the defaults to 50ms same as V1 after surface area review" This reverts commit e7d2295. * Bytes Based batching for SendMessageRequest Batching (#5540) * Initial changes * Initial changes 2 * Byte Based batching for SendMessage API * Byte Based batching for SendMessage API * Handled comments * Checkstyle issue * Add User Agent for Sqs Calls made using Automatic Batching Manager (#5546) * Add User Agent for Sqs Calls made using Automatic Batching Manager as hll/abm * Review comments * Update comments from PR 5488 (#5550) * Update comments of PR 5488 * Update comments from PR 5488 * Handled surface area review comments (#5563) * Initial version * Intermediate changes * Update after internal poll * ResponseCOnfiguration construction updated * RequestOverride configuration check added to Bypass batch manager * Handled review comments * Removed TODO since validations are handled in BatchPverrideConfiguration * Fix issue where the Scheduled Timeout was incorrectly completing the futures with empty messages (#5571) * Fix issue where the Scheduled Timeout was incorrectly completing the futures with empty messages * Handled review comments * Integ test for Automatic Request Batching (#5576) * feat(sqs): add BatchManager for client-side request batching to Amazon SQS The new BatchManager allows for simple request batching using client-side buffering, improving cost efficiency and reducing the number of requests sent to Amazon SQS. The client-side buffering supports up to 10 requests per batch and is supported by the SqsAsyncClient. Batched requests, along with receive message polling, help to increase throughput. * Add check for scheduledExecutor such that it not null when creating SqsAsyncBatchManager (#5582) * Add check for scheduledExecutor such that it not null when creating SqsAsyncBatchManager * Update services/sqs/src/test/java/software/amazon/awssdk/services/sqs/batchmanager/SqsAsyncBatchManagerBuilderTest.java Co-authored-by: David Ho <[email protected]> * Update services/sqs/src/test/java/software/amazon/awssdk/services/sqs/batchmanager/SqsAsyncBatchManagerBuilderTest.java Co-authored-by: David Ho <[email protected]> * Update services/sqs/src/test/java/software/amazon/awssdk/services/sqs/batchmanager/SqsAsyncBatchManagerBuilderTest.java Co-authored-by: David Ho <[email protected]> --------- Co-authored-by: David Ho <[email protected]> * Updating Timeouts in gets so that we dont wait infinitely --------- Co-authored-by: David Ho <[email protected]>

joviegas added 5 commits August 6, 2024 15:29

Add Recieve Buffer Queue And its related configuration

b8734a2

Merge branch 'feature/master/sqs-batch-manager' into joviegas/receive…

ca6cf94

…_buffer_queue

Update ReceiveBatch manager

ed57320

Recieve Batch Manager Implementation

79b93ec

Receive Batch Manager Implemetation

6ff4a77

joviegas requested a review from a team as a code owner August 8, 2024 19:53

dagnir reviewed Aug 8, 2024

View reviewed changes

Handled review comments

6e47217

joviegas force-pushed the joviegas/receive_buffer_queue branch from 5a887a7 to 6e47217 Compare August 14, 2024 22:56

joviegas added 3 commits August 15, 2024 10:34

Checkstyle failure

766fbef

Flsky test case fixed

8a0fb75

Flaky test case fixed

18d7af2

zoewangg reviewed Aug 15, 2024

View reviewed changes

Hamdled review comments

ef0d98c

zoewangg reviewed Aug 16, 2024

View reviewed changes

joviegas added 4 commits August 17, 2024 13:01

Handled comments

61db2c6

Merge branch 'feature/master/sqs-batch-manager' into joviegas/receive…

239c4f1

…_buffer_queue

Removed ReceiveMessageCompletableFuture

f510b82

SdkClosable implemented

1a0b987

joviegas force-pushed the joviegas/receive_buffer_queue branch from 93d2b7e to 1a0b987 Compare August 18, 2024 01:00

joviegas added 3 commits August 19, 2024 17:07

Added ReceiveMessageBatchManager class for completeness

e0f7b34

Checkstyle issues

87eef9f

Null checks

ac17c3d

zoewangg reviewed Aug 22, 2024

View reviewed changes

zoewangg reviewed Aug 23, 2024

View reviewed changes

Handled comments from Zoe

f6031c7

joviegas merged commit 18c06cc into feature/master/sqs-batch-manager Aug 27, 2024
16 of 17 checks passed

zoewangg reviewed Sep 3, 2024

View reviewed changes

joviegas mentioned this pull request Sep 3, 2024

Update comments from PR 5488 #5550

Merged

zoewangg reviewed Sep 3, 2024

View reviewed changes

joviegas deleted the joviegas/receive_buffer_queue branch January 15, 2025 17:36

Receive Batch Manager Implementation #5488

Receive Batch Manager Implementation #5488

Uh oh!

Conversation

joviegas commented Aug 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation and Context

Modifications

Design Details

Testing

License

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dagnir commented Aug 8, 2024

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zoewangg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zoewangg left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

joviegas commented Aug 8, 2024 •

edited

Loading

zoewangg left a comment •

edited

Loading