Add support for SqsTemplate to send more than 10 messages at once

docs/src/main/asciidoc/sqs.adoc

@@ -319,8 +319,10 @@ SendResult.Batch<T> sendMany(String queue, Collection<Message<T>> messages);
 ```
 
 NOTE: To send a collection of objects, it is recommended to use `sendMany(String queue, Collection<Message<T>> messages)` to optimize throughput.
-To send a collection of objects in a single message, the collection must be wrapped in an object. The underlying AWS SQS API has a limitation that only up to 10 messages can be sent in a single batch request using `SendMessageBatch`. If more than 10 messages are passed to `sendMany()`, the AWS SDK will throw a `TooManyEntriesInBatchRequestException`.
-This limitation is documented in the https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessageBatch.html[AWS SQS API Reference for SendMessageBatch]. As of now, Spring Cloud AWS does not automatically split larger collections into smaller batches of 10 or fewer messages. Users are responsible for ensuring the batch size complies with this AWS limit.
+To send a collection of objects in a single message, the collection must be wrapped in an object.
+The underlying AWS SQS API has a limitation that only up to 10 messages can be sent in a single batch request using `SendMessageBatch`.
+Since 4.2.0, Spring Cloud AWS automatically splits larger collections into batches of 10 or fewer messages.
+For standard queues, all batches are sent in parallel. For FIFO queues, batches are sent in parallel, grouped by message group ID.
 
 An example using the `options` variant follows:
 
@@ -339,6 +341,28 @@ If `messageDeduplicationId` is not provided and content deduplication is disable
 The generated values can be retrieved in the headers of the `Message` contained in the `SendResult` object.
 
 
+===== Batch sending behavior
+
+The `sendMany` method handles collections of messages transparently, with no behavior changes for batches of 10 or fewer messages.
+Larger collections are automatically partitioned into batches of up to 10 messages, the maximum supported by the SQS `SendMessageBatch` API.
+
+For standard queues, all partitions are sent in parallel.
+
+For FIFO queues, messages are first grouped by message group ID.
+Each group is kept whole: groups larger than 10 messages are sent sequentially to preserve the message order within the group, while groups of up to 10 messages are packed into shared batches that are sent in parallel.
+This preserves the ordering guarantees that FIFO queues provide per message group.
+
+Send failures within a batch can occur in the following ways:
+
+* **Partial failures**: If only some messages in a single batch request fail, the successful ones are returned in `successful` and the failed ones in `failed` as `SendResult.Failed` entries in the `SendResult.Batch`.
+* **Skipped batches (FIFO only)**: If a batch fails within a FIFO group, the remaining batches for that group are not sent and are returned as failed entries with the message `Skipped due to previous batch failure`.
+* **Batch-level exceptions**: If sending a batch fails entirely with an exception, all messages in that batch are returned as failed entries, with the exception available in the `additionalInformation` map under the `SqsTemplateParameters.EXCEPTION_PARAMETER_NAME` key.
+
+How failed entries are surfaced depends on the `sendBatchFailureHandlingStrategy` option.
+By default (`THROW`), a `SendBatchOperationFailedException` is thrown if at least one message fails, containing the `SendResult.Batch` with both successful and failed messages.
+With `DO_NOT_THROW`, no exception is thrown and the `SendResult.Batch` is returned directly.
+
+
 ===== SendResult
 
 The `SendResult` record contains useful information on the send operation.

spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SendManyBatchSample.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.sqs.sample;
+
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import io.awspring.cloud.sqs.operations.SendResult;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import java.util.List;
+import java.util.stream.IntStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Sample demonstrating {@link SqsTemplate#sendMany} sending more than 10 messages at once. The template automatically
+ * partitions the messages into batches of 10 and sends them in parallel (for standard queues) or sequentially per
+ * message group (for FIFO queues).
+ *
+ * @author José Iêdo
+ */
+@Configuration
+public class SendManyBatchSample {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(SendManyBatchSample.class);
+
+	private static final String QUEUE_NAME = "send-many-batch-queue";
+
+	@SqsListener(queueNames = QUEUE_NAME, maxMessagesPerPoll = "25", maxConcurrentMessages = "25")
+	void listen(List<Message<String>> messages) {
+		LOGGER.info("Received {} messages: {}", messages.size(), messages.stream().map(Message::getPayload).toList());
+	}
+
+	@Bean
+	public ApplicationRunner sendManyMessages(SqsTemplate sqsTemplate) {
+		return args -> {
+			List<Message<String>> messages = IntStream.range(0, 25).mapToObj(index -> "Message-" + index)
+					.map(payload -> MessageBuilder.withPayload(payload).build()).toList();
+			LOGGER.info("Sending {} messages to queue {}", messages.size(), QUEUE_NAME);
+			SendResult.Batch<String> result = sqsTemplate.sendMany(QUEUE_NAME, messages);
+			LOGGER.info("Sent successfully: {}, failed: {}", result.successful().size(), result.failed().size());
+		};
+	}
+
+}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/SqsTemplate.java

@@ -16,6 +16,7 @@
 package io.awspring.cloud.sqs.operations;
 
 import io.awspring.cloud.core.support.JacksonPresent;
+import io.awspring.cloud.sqs.CollectionUtils;
 import io.awspring.cloud.sqs.FifoUtils;
 import io.awspring.cloud.sqs.MessageHeaderUtils;
 import io.awspring.cloud.sqs.QueueAttributesResolver;
@@ -35,9 +36,11 @@
 import io.awspring.cloud.sqs.support.converter.legacy.LegacyJackson2SqsMessagingMessageConverter;
 import io.awspring.cloud.sqs.support.observation.SqsTemplateObservation;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
@@ -80,11 +83,14 @@
  * @author Zhong Xi Lu
  * @author Hyunggeol Lee
  * @author Jeongmin Kim
+ * @author José Iêdo
  *
  * @since 3.0
  */
 public class SqsTemplate extends AbstractMessagingTemplate<Message> implements SqsOperations, SqsAsyncOperations {
 
+	private static final int SQS_MAX_BATCH_SIZE = 10;
+
 	private static final Logger logger = LoggerFactory.getLogger(SqsTemplate.class);
 
 	private static final SqsTemplateObservation.SqsSpecifics SQS_OBSERVATION_SPECIFICS = new SqsTemplateObservation.SqsSpecifics();
@@ -365,13 +371,161 @@ private SendMessageRequest doCreateSendMessageRequest(Message message, QueueAttr
 				.messageSystemAttributes(mapMessageSystemAttributes(message)).build();
 	}
 
+	/**
+	 * Sends a collection of messages using one or more SQS batch requests.
+	 * <p>
+	 * The provided messages are automatically partitioned into batches of up to 10 messages, which is the maximum size
+	 * supported by Amazon SQS.
+	 * <p>
+	 * For standard queues, all batches are sent in parallel.
+	 * <p>
+	 * For FIFO queues, messages are first grouped by
+	 * {@link io.awspring.cloud.sqs.listener.SqsHeaders.MessageSystemAttributes#SQS_MESSAGE_GROUP_ID_HEADER message
+	 * group ID}. Groups larger than 10 messages are sent sequentially to preserve message ordering within each group,
+	 * with a skip-on-failure strategy: if a batch completes with a partial failure, no subsequent batches for that
+	 * group are sent.
+	 * <p>
+	 * Groups with up to 10 messages are bin-packed into shared batches on a best-effort basis (first-fit decreasing),
+	 * reducing the number of requests while keeping each group whole within a single batch to preserve ordering. Packed
+	 * batches are sent in parallel, as are large-group chains across different groups.
+	 */
 	@Override
 	protected <T> CompletableFuture<SendResult.Batch<T>> doSendBatchAsync(String endpointName,
 			Collection<Message> messages, Collection<org.springframework.messaging.Message<T>> originalMessages) {
 		logger.debug("Sending messages {} to endpoint {}", messages, endpointName);
+		Map<String, org.springframework.messaging.Message<T>> originalMessagesById = originalMessages.stream()
+				.collect(Collectors.toMap(MessageHeaderUtils::getRawMessageId, msg -> msg));
+		if (messages.size() <= SQS_MAX_BATCH_SIZE) {
+			return sendSingleBatch(endpointName, messages, originalMessagesById);
+		}
+		return FifoUtils.isFifo(endpointName) ? sendFifoBatches(endpointName, messages, originalMessagesById)
+				: sendStandardBatches(endpointName, messages, originalMessagesById);
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> sendSingleBatch(String endpointName,
+			Collection<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
 		return createSendMessageBatchRequest(endpointName, messages).thenCompose(this.sqsAsyncClient::sendMessageBatch)
-				.thenApply(response -> createSendResultBatch(response, endpointName, originalMessages.stream()
-						.collect(Collectors.toMap(MessageHeaderUtils::getRawMessageId, msg -> msg))));
+				.thenApply(response -> createSendResultBatch(response, endpointName, originalMessagesById));
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> sendPartitionedBatch(String endpointName,
+			Collection<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		return sendSingleBatch(endpointName, messages, originalMessagesById)
+				.exceptionally(t -> createFailedBatchResult(messages, t, endpointName, originalMessagesById));
+	}
+
+	private <T> SendResult.Batch<T> createFailedBatchResult(Collection<Message> partition, Throwable throwable,
+			String endpointName, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		Throwable cause = throwable;
+		if (cause instanceof java.util.concurrent.CompletionException completionException
+				&& completionException.getCause() != null) {
+			cause = completionException.getCause();
+		}
+		Throwable exceptionParameter = cause != null ? cause : throwable;
+		Map<String, Object> additionalInformation = Map.of(SqsTemplateParameters.EXCEPTION_PARAMETER_NAME,
+				exceptionParameter);
+		String errorMessage = cause != null && cause.getMessage() != null ? cause.getMessage() : "Unknown error";
+		List<SendResult.Failed<T>> failed = partition.stream().map(msg -> new SendResult.Failed<>(errorMessage,
+				endpointName, originalMessagesById.get(msg.messageId()), additionalInformation)).toList();
+		return new SendResult.Batch<>(List.of(), failed);
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> sendStandardBatches(String endpointName,
+			Collection<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		List<CompletableFuture<SendResult.Batch<T>>> futures = CollectionUtils.partition(messages, SQS_MAX_BATCH_SIZE)
+				.stream().map(partition -> sendPartitionedBatch(endpointName, partition, originalMessagesById))
+				.toList();
+		return combineBatchFutures(futures);
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> sendFifoBatches(String endpointName,
+			Collection<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		Map<String, List<Message>> groupedByMessageGroup = messages.stream().collect(Collectors.groupingBy(msg -> {
+			String groupId = msg.attributes().get(MessageSystemAttributeName.MESSAGE_GROUP_ID);
+			return groupId != null ? groupId : "";
+		}));
+		Map<Boolean, List<List<Message>>> partitioned = groupedByMessageGroup.values().stream()
+				.collect(Collectors.partitioningBy(group -> group.size() <= SQS_MAX_BATCH_SIZE));
+		List<List<Message>> smallGroups = partitioned.get(true);
+		List<List<Message>> largeGroups = partitioned.get(false);
+		List<CompletableFuture<SendResult.Batch<T>>> futures = largeGroups.stream()
+				.map(msgs -> sendSequentialBatches(endpointName, msgs, originalMessagesById))
+				.collect(Collectors.toList());
+		if (!smallGroups.isEmpty()) {
+			binPackSmallFifoGroups(smallGroups, SQS_MAX_BATCH_SIZE).stream()
+					.map(batch -> sendPartitionedBatch(endpointName, batch, originalMessagesById))
+					.forEach(futures::add);
+		}
+		return combineBatchFutures(futures);
+	}
+
+	/**
+	 * Bin-pack small FIFO groups into shared batches using first-fit decreasing algorithm. Each group is kept whole
+	 * within a single batch. Groups are sorted by size descending before packing to minimize the number of batches.
+	 * @param smallGroups groups with size <= maxBatchSize
+	 * @param maxBatchSize the maximum number of messages per batch (SQS limit is 10)
+	 * @return packed batches, each containing one or more whole groups
+	 */
+	private static List<List<Message>> binPackSmallFifoGroups(List<List<Message>> smallGroups, int maxBatchSize) {
+		Assert.notNull(smallGroups, "smallGroups must not be null");
+		Assert.isTrue(maxBatchSize > 0, "maxBatchSize must be positive");
+		smallGroups.sort((a, b) -> Integer.compare(b.size(), a.size()));
+		List<List<Message>> packedBatches = new ArrayList<>();
+		for (List<Message> group : smallGroups) {
+			boolean packed = false;
+			for (List<Message> batch : packedBatches) {
+				if (batch.size() + group.size() <= maxBatchSize) {
+					batch.addAll(group);
+					packed = true;
+					break;
+				}
+			}
+			if (!packed) {
+				packedBatches.add(new ArrayList<>(group));
+			}
+		}
+		return packedBatches;
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> combineBatchFutures(
+			List<CompletableFuture<SendResult.Batch<T>>> futures) {
+		return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
+				.thenApply(v -> futures.stream().map(CompletableFuture::join)
+						.reduce(new SendResult.Batch<>(List.of(), List.of()), this::mergeBatchResults));
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> sendSequentialBatches(String endpointName,
+			List<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		CompletableFuture<SendResult.Batch<T>> result = CompletableFuture
+				.completedFuture(new SendResult.Batch<>(List.of(), List.of()));
+		for (Collection<Message> partition : CollectionUtils.partition(messages, SQS_MAX_BATCH_SIZE)) {
+			result = result.thenCompose(acc -> {
+				if (!acc.failed().isEmpty()) {
+					return CompletableFuture.completedFuture(
+							mergeBatchResults(acc, createSkippedResult(partition, endpointName, originalMessagesById)));
+				}
+				return sendPartitionedBatch(endpointName, partition, originalMessagesById)
+						.thenApply(batchResult -> mergeBatchResults(acc, batchResult));
+			});
+		}
+		return result;
+	}
+
+	private <T> SendResult.Batch<T> createSkippedResult(Collection<Message> partition, String endpointName,
+			Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		List<SendResult.Failed<T>> skipped = partition.stream()
+				.map(msg -> new SendResult.Failed<>("Skipped due to previous batch failure", endpointName,
+						originalMessagesById.get(msg.messageId()), Map.of()))
+				.toList();
+		return new SendResult.Batch<>(List.of(), skipped);
+	}
+
+	private <T> SendResult.Batch<T> mergeBatchResults(SendResult.Batch<T> batch1, SendResult.Batch<T> batch2) {
+		List<SendResult<T>> allSuccessful = new ArrayList<>(batch1.successful());
+		allSuccessful.addAll(batch2.successful());
+		List<SendResult.Failed<T>> allFailed = new ArrayList<>(batch1.failed());
+		allFailed.addAll(batch2.failed());
+		return new SendResult.Batch<>(allSuccessful, allFailed);
 	}
 
 	private <T> SendResult.Batch<T> createSendResultBatch(SendMessageBatchResponse response, String endpointName,
@@ -937,8 +1091,8 @@ public SqsReceiveOptionsImpl visibilityTimeout(Duration visibilityTimeout) {
 		@Override
 		public SqsReceiveOptionsImpl maxNumberOfMessages(Integer maxNumberOfMessages) {
 			Assert.notNull(maxNumberOfMessages, "maxNumberOfMessages must not be null");
-			Assert.isTrue(maxNumberOfMessages > 0 && maxNumberOfMessages <= 10,
-					"maxNumberOfMessages must be between 0 and 10");
+			Assert.isTrue(maxNumberOfMessages > 0 && maxNumberOfMessages <= SQS_MAX_BATCH_SIZE,
+					"maxNumberOfMessages must be between 0 and " + SQS_MAX_BATCH_SIZE);
 			this.maxNumberOfMessages = maxNumberOfMessages;
 			return this;
 		}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/SqsTemplateParameters.java

@@ -44,4 +44,9 @@ public class SqsTemplateParameters {
 	 */
 	public static final String ERROR_CODE_PARAMETER_NAME = "code";
 
+	/**
+	 * The exception that was thrown.
+	 */
+	public static final String EXCEPTION_PARAMETER_NAME = "exception";
+
 }