Changes in automatic Amazon SQS request batching from version 1 to version 2 - Amazon SDK for Java 2.x
High-level changesUsing automatic SQS request batching Asynchronous return type differencesAsynchronous completion handling differencesKey configuration parameters
Services or capabilities described in Amazon Web Services documentation might vary by Region. To see the differences applicable to the China Regions, see Getting Started with Amazon Web Services in China (PDF).

Changes in automatic Amazon SQS request batching from version 1 to version 2

This topic details the changes in automatic request batching for Amazon SQS between version 1 and version 2 of the Amazon SDK for Java.

High-level changes

The Amazon SDK for Java 1.x performs client-side buffering using a separate AmazonSQSBufferedAsyncClient class that requires explicit initialization for request batching.

The Amazon SDK for Java 2.x simplifies and enhances buffering functionality with the SqsAsyncBatchManager. The implementation of this interface provides automatic request batching capabilities directly integrated with the standard SqsAsyncClient. To learn about v2's SqsAsyncBatchManager, see the Use automatic request batching for Amazon SQS with the Amazon SDK for Java 2.x topic in this guide.

Change v1 v2

Maven dependencies

 
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk-bom</artifactId>
            <version>1.12.7821</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-java-sdk-sqs</artifactId>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>software.amazon.awssdk</groupId>
            <artifactId>bom</artifactId>
            <version>2.31.152</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <dependency>
        <groupId>software.amazon.awssdk</groupId>
        <artifactId>sqs</artifactId>
    </dependency>
</dependencies>
Package names com.amazonaws.services.sqs.buffered software.amazon.awssdk.services.sqs.batchmanager
Class names

AmazonSQSBufferedAsyncClient

 SqsAsyncBatchManager

1 Latest version. 2 Latest version.

Using automatic SQS request batching

Change v1 v2
Create a batch manager 
AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient();
AmazonSQSAsync bufferedSqs = new 
            AmazonSQSBufferedAsyncClient(sqsAsync);
 
SqsAsyncClient asyncClient = SqsAsyncClient.create();
SqsAsyncBatchManager sqsAsyncBatchManager = 
            asyncClient.batchManager();
Create a batch manager with custom configuration 
AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient();

QueueBufferConfig queueBufferConfig = new QueueBufferConfig()
        .withMaxBatchOpenMs(200)
        .withMaxBatchSize(10)
        .withMinReceiveWaitTimeMs(1000)
        .withVisibilityTimeoutSeconds(20)
        .withReceiveMessageAttributeNames(messageAttributeValues);

AmazonSQSAsync bufferedSqs = 
        new AmazonSQSBufferedAsyncClient(sqsAsync, queueBufferConfig);
 
BatchOverrideConfiguration batchOverrideConfiguration = 
    BatchOverrideConfiguration.builder()
        .sendRequestFrequency(Duration.ofMillis(200))
        .maxBatchSize(10)
        .receiveMessageMinWaitDuration(Duration.ofMillis(1000))
        .receiveMessageVisibilityTimeout(Duration.ofSeconds(20))
        .receiveMessageSystemAttributeNames(messageSystemAttributeNames)
        .receiveMessageAttributeNames(messageAttributeValues)
        .build();

SqsAsyncBatchManager sqsAsyncBatchManager = SqsAsyncBatchManager.builder()
        .overrideConfiguration(batchOverrideConfiguration)
        .client(SqsAsyncClient.create())
        .scheduledExecutor(Executors.newScheduledThreadPool(8))
        .build();
Send messages 
Future<SendMessageResult> sendResultFuture = 
        bufferedSqs.sendMessageAsync(new SendMessageRequest()
                .withQueueUrl(queueUrl)
                .withMessageBody(body));
 
CompletableFuture<SendMessageResponse> sendCompletableFuture = 
        sqsAsyncBatchManager.sendMessage(
                SendMessageRequest.builder()
                        .queueUrl(queueUrl)
                        .messageBody(body)
                        .build());
Delete messages 
Future<DeleteMessageResult> deletResultFuture =
        bufferedSqs.deleteMessageAsync(new DeleteMessageRequest()
                .withQueueUrl(queueUrl));
 
CompletableFuture<DeleteMessageResponse> deleteResultCompletableFuture
        = sqsAsyncBatchManager.deleteMessage(
                DeleteMessageRequest.builder()
                        .queueUrl(queueUrl)
                        .build());
Change visibility of messages 
Future<ChangeMessageVisibilityResult> changeVisibilityResultFuture =
        bufferedSqs.changeMessageVisibilityAsync
                (new ChangeMessageVisibilityRequest()
                        .withQueueUrl(queueUrl)
                        .withVisibilityTimeout(20));
 
CompletableFuture<ChangeMessageVisibilityResponse> changeResponseCompletableFuture
        = sqsAsyncBatchManager.changeMessageVisibility(
                ChangeMessageVisibilityRequest.builder()
                        .queueUrl(queueUrl)
                        .visibilityTimeout(20)
                        .build());
Receive messages 
ReceiveMessageResult receiveResult =
        bufferedSqs.receiveMessage(
                new ReceiveMessageRequest()
                        .withQueueUrl(queueUrl));
 
CompletableFuture<ReceiveMessageResponse> 
        responseCompletableFuture = sqsAsyncBatchManager.receiveMessage(
                ReceiveMessageRequest.builder()
                        .queueUrl(queueUrl)
                        .build());

Asynchronous return type differences

Change v1 v2
Return type Future<ResultType> CompletableFuture<ResponseType>
Callback mechanism Requires an AsyncHandler with separate onSuccess and onError methods Uses CompletableFuture APIs provided by the JDK, such as whenComplete(), thenCompose(), thenApply()
Exception handling Uses AsyncHandler#onError() method Uses CompletableFuture APIs provided by the JDK, such as exceptionally(), handle(), or whenComplete()
Cancellation Basic support through Future.cancel() Cancelling a parent CompletableFuture automatically cancels all dependent futures in the chain

Asynchronous completion handling differences

Change v1 v2
Response handler implementation 
Future<ReceiveMessageResult> future = bufferedSqs.receiveMessageAsync(
        receiveRequest,
        new AsyncHandler<ReceiveMessageRequest, ReceiveMessageResult>() {
            @Override
            public void onSuccess(ReceiveMessageRequest request, 
                              ReceiveMessageResult result) {
                List<Message> messages = result.getMessages();
                System.out.println("Received " + messages.size() + " messages");
                for (Message message : messages) {
                    System.out.println("Message ID: " + message.getMessageId());
                    System.out.println("Body: " + message.getBody());
                }
            }

            @Override
            public void onError(Exception e) {
                System.err.println("Error receiving messages: " + e.getMessage());
                e.printStackTrace();
            }
        }
);
 
CompletableFuture<ReceiveMessageResponse> completableFuture = sqsAsyncBatchManager
               .receiveMessage(ReceiveMessageRequest.builder()
               .queueUrl(queueUrl).build())
        .whenComplete((receiveMessageResponse, throwable) -> {
            if (throwable != null) {
                System.err.println("Error receiving messages: " + throwable.getMessage());
                throwable.printStackTrace();
            } else {
                List<Message> messages = receiveMessageResponse.messages();
                System.out.println("Received " + messages.size() + " messages");
                for (Message message : messages) {
                    System.out.println("Message ID: " + message.messageId());
                    System.out.println("Body: " + message.body());
                }
            }
        });

Key configuration parameters

Parameter v1 v2
Maximum batch size maxBatchSize (default 10 requests per batch) maxBatchSize (default 10 requests per batch)
Batch wait time maxBatchOpenMs (default 200 ms) sendRequestFrequency (default 200 ms)
Visibility timeout visibilityTimeoutSeconds (-1 for queue default) receiveMessageVisibilityTimeout (queue default)
Minimum wait time longPollWaitTimeoutSeconds (20s when longPoll is true) receiveMessageMinWaitDuration (default 50 ms)
Message attributes Set using ReceiveMessageRequest receiveMessageAttributeNames (none by default)
System attributes Set using ReceiveMessageRequest receiveMessageSystemAttributeNames (none by default)
Long polling longPoll (default is true) Not supported to avoid open connections waiting until the server sends the messages
Maximum wait time for long polling longPollWaitTimeoutSeconds (default 20s) Not supported to avoid open connections waiting until the server sends the messages
Maximum number of prefetched receive batches stored client-side maxDoneReceiveBatches (10 batches) Not supported because it is handled internally
Maximum number of active outbound batches processed simultaneously maxInflightOutboundBatches (default 5 batches) Not supported because it is handled internally
Maximum number of active receive batches processed simultaneously maxInflightReceiveBatches (default 10 batches) Not supported because it is handled internally