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/CollectionUtils.java

@@ -25,9 +25,8 @@ public class CollectionUtils {
 	public static <T> List<Collection<T>> partition(Collection<T> messagesToAck, int pageSize) {
 		List<T> messagesToUse = getAsList(messagesToAck);
 		int totalSize = messagesToUse.size();
-		return IntStream.rangeClosed(0, (totalSize - 1) / pageSize)
-				.mapToObj(index -> messagesToUse.subList(index * pageSize, Math.min((index + 1) * pageSize, totalSize)))
-				.toList();
+		return IntStream.rangeClosed(0, (totalSize - 1) / pageSize).mapToObj(index -> (Collection<T>) messagesToUse
+				.subList(index * pageSize, Math.min((index + 1) * pageSize, totalSize))).toList();
 	}
 
 	private static <T> List<T> getAsList(Collection<T> elements) {

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,137 @@ 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>> 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 -> 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 : "";
+		}));
+		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 -> sendSingleBatch(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
+	 */
+	protected 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 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,
@@ -937,8 +1067,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/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";