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

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,6 +83,7 @@
  * @author Zhong Xi Lu
  * @author Hyunggeol Lee
  * @author Jeongmin Kim
+ * @author José Iêdo
  *
  * @since 3.0
  */
@@ -365,13 +369,98 @@ 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}. Batches belonging to the same message group are sent sequentially to preserve message ordering, while
+	 * different groups may be processed concurrently.
+	 * <p>
+	 * If a batch for a FIFO message group completes with a partial failure, no subsequent batches for that group are
+	 * sent, ensuring the ordering guarantees of the message group are maintained.
+	 */
 	@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() <= 10) {
+			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>> sendStandardBatches(String endpointName,
+			Collection<Message> messages, Map<String, org.springframework.messaging.Message<T>> originalMessagesById) {
+		List<CompletableFuture<SendResult.Batch<T>>> futures = CollectionUtils.partition(messages, 10).stream()
+				.map(partition -> sendSingleBatch(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 : "";
+		}));
+		List<CompletableFuture<SendResult.Batch<T>>> groupFutures = groupedByMessageGroup.values().stream()
+				.map(msgs -> sendSequentialBatches(endpointName, msgs, originalMessagesById)).toList();
+		return combineBatchFutures(groupFutures);
+	}
+
+	private <T> CompletableFuture<SendResult.Batch<T>> combineBatchFutures(
+			List<CompletableFuture<SendResult.Batch<T>>> futures) {
+		return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+				.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, 10)) {
+			result = result.thenCompose(acc -> {
+				if (!acc.failed().isEmpty()) {
+					return CompletableFuture.completedFuture(
+							mergeBatchResults(acc, createSkippedResult(partition, endpointName, originalMessagesById)));
+				}
+				return sendSingleBatch(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,

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsFifoIntegrationTests.java

@@ -93,6 +93,7 @@
  *
  * @author Tomaz Fernandes
  * @author Mikhail Strokov
+ * @author José Iêdo
  */
 @SpringBootTest
 class SqsFifoIntegrationTests extends BaseSqsIntegrationTest {
@@ -123,6 +124,10 @@ class SqsFifoIntegrationTests extends BaseSqsIntegrationTest {
 
 	static final String FIFO_VISIBILITY_TIMEOUT_EXTENSION_QUEUE_NAME = "fifo_visibility_timeout_extension_test_queue.fifo";
 
+	static final String FIFO_SEND_MORE_THAN_10_SINGLE_GROUP_QUEUE_NAME = "fifo_send_more_than_10_single_group.fifo";
+
+	static final String FIFO_SEND_MORE_THAN_10_MULTIPLE_GROUPS_QUEUE_NAME = "fifo_send_more_than_10_multiple_groups.fifo";
+
 	private static final String ERROR_ON_ACK_FACTORY = "errorOnAckFactory";
 	private static final String VISIBILITY_TIMEOUT_EXTENSION_FACTORY = "visibilityTimeoutExtensionFactory";
 
@@ -174,6 +179,8 @@ static void beforeTests() {
 				createFifoQueue(client, FIFO_MANUALLY_CREATE_BATCH_CONTAINER_QUEUE_NAME),
 				createFifoQueue(client, OBSERVES_MESSAGE_FIFO_QUEUE_NAME),
 				createFifoQueue(client, FIFO_VISIBILITY_TIMEOUT_EXTENSION_QUEUE_NAME, getVisibilityAttribute("5")),
+				createFifoQueue(client, FIFO_SEND_MORE_THAN_10_SINGLE_GROUP_QUEUE_NAME),
+				createFifoQueue(client, FIFO_SEND_MORE_THAN_10_MULTIPLE_GROUPS_QUEUE_NAME),
 				createFifoQueue(client, FIFO_MANUALLY_CREATE_BATCH_FACTORY_QUEUE_NAME)).join();
 	}
 
@@ -535,6 +542,34 @@ void manuallyCreatesBatchFactory() throws Exception {
 		assertThat(messagesContainer.manuallyCreatedBatchFactoryMessages).containsExactlyElementsOf(values);
 	}
 
+	@Test
+	void shouldSendMoreThan10FifoMessagesInSingleGroup() {
+		String messageGroupId = UUID.randomUUID().toString();
+		List<Message<String>> messages = IntStream.range(0, 25)
+				.mapToObj(i -> createMessage("payload-" + i, messageGroupId)).toList();
+		SqsTemplate fifoTemplate = SqsTemplate.newTemplate(createAsyncClient());
+		SendResult.Batch<String> result = fifoTemplate.sendMany(FIFO_SEND_MORE_THAN_10_SINGLE_GROUP_QUEUE_NAME,
+				messages);
+		assertThat(result.successful()).hasSize(25);
+		assertThat(result.failed()).isEmpty();
+	}
+
+	@Test
+	void shouldSendMoreThan10FifoMessagesAcrossMultipleGroups() {
+		List<String> valuesGroup1 = IntStream.range(0, 20).mapToObj(String::valueOf).collect(toList());
+		List<String> valuesGroup2 = IntStream.range(0, 15).mapToObj(String::valueOf).collect(toList());
+		String group1 = UUID.randomUUID().toString();
+		String group2 = UUID.randomUUID().toString();
+		List<Message<String>> messages = new ArrayList<>();
+		messages.addAll(createMessagesFromValues(group1, valuesGroup1));
+		messages.addAll(createMessagesFromValues(group2, valuesGroup2));
+		SqsTemplate fifoTemplate = SqsTemplate.newTemplate(createAsyncClient());
+		SendResult.Batch<String> result = fifoTemplate.sendMany(FIFO_SEND_MORE_THAN_10_MULTIPLE_GROUPS_QUEUE_NAME,
+				messages);
+		assertThat(result.successful()).hasSize(35);
+		assertThat(result.failed()).isEmpty();
+	}
+
 	private Message<String> createMessage(String body, String messageGroupId) {
 		return MessageBuilder.withPayload(body)
 				.setHeader(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, messageGroupId)

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java

@@ -342,6 +342,15 @@ void receivesMessageBatch() throws Exception {
 		assertThat(latchContainer.acknowledgementCallbackBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
+	@Test
+	void shouldSendMoreThan10MessagesAtOnce() {
+		List<Message<String>> messages = IntStream.range(0, 25)
+				.mapToObj(i -> MessageBuilder.withPayload("moreThan10-payload-" + i).build()).toList();
+		SendResult.Batch<String> result = sqsTemplate.sendMany(RECEIVES_MESSAGE_QUEUE_NAME, messages);
+		assertThat(result.successful()).hasSize(25);
+		assertThat(result.failed()).isEmpty();
+	}
+
 	@Test
 	void receivesMessageAsync() throws Exception {
 		String messageBody = "receivesMessageAsync-payload";