Skip to content

Commit e1c50e1

Browse files
joseiedojoseiedo
committed
cleanup
1 parent f4169ec commit e1c50e1

3 files changed

Lines changed: 66 additions & 115 deletions

File tree

‎spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/CollectionUtils.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,8 @@ public class CollectionUtils {
2525
public static <T> List<Collection<T>> partition(Collection<T> messagesToAck, int pageSize) {
2626
List<T> messagesToUse = getAsList(messagesToAck);
2727
int totalSize = messagesToUse.size();
28-
return IntStream.rangeClosed(0, (totalSize - 1) / pageSize)
29-
.mapToObj(index -> (Collection<T>) messagesToUse.subList(index * pageSize, Math.min((index + 1) * pageSize, totalSize)))
30-
.toList();
28+
return IntStream.rangeClosed(0, (totalSize - 1) / pageSize).mapToObj(index -> (Collection<T>) messagesToUse
29+
.subList(index * pageSize, Math.min((index + 1) * pageSize, totalSize))).toList();
3130
}
3231

3332
private static <T> List<T> getAsList(Collection<T> elements) {

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

Lines changed: 46 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -421,33 +421,56 @@ private <T> CompletableFuture<SendResult.Batch<T>> sendFifoBatches(String endpoi
421421
String groupId = msg.attributes().get(MessageSystemAttributeName.MESSAGE_GROUP_ID);
422422
return groupId != null ? groupId : "";
423423
}));
424-
Map<Boolean, List<List<Message>>> partitioned = groupedByMessageGroup.values().stream()
425-
.collect(Collectors.partitioningBy(group -> group.size() <= SQS_MAX_BATCH_SIZE));
426-
List<List<Message>> smallGroups = partitioned.get(true);
427-
List<List<Message>> largeGroups = partitioned.get(false);
428-
List<CompletableFuture<SendResult.Batch<T>>> futures = largeGroups.stream()
429-
.map(msgs -> sendSequentialBatches(endpointName, msgs, originalMessagesById))
430-
.collect(Collectors.toList());
424+
List<List<Message>> largeGroups = new ArrayList<>();
425+
List<List<Message>> fullSizeGroups = new ArrayList<>();
426+
List<List<Message>> smallGroups = new ArrayList<>();
427+
for (List<Message> group : groupedByMessageGroup.values()) {
428+
if (group.size() > SQS_MAX_BATCH_SIZE) {
429+
largeGroups.add(group);
430+
}
431+
else if (group.size() == SQS_MAX_BATCH_SIZE) {
432+
fullSizeGroups.add(group);
433+
}
434+
else {
435+
smallGroups.add(group);
436+
}
437+
}
438+
List<CompletableFuture<SendResult.Batch<T>>> futures = new ArrayList<>();
439+
largeGroups.stream().map(msgs -> sendSequentialBatches(endpointName, msgs, originalMessagesById))
440+
.forEach(futures::add);
441+
fullSizeGroups.stream().map(batch -> sendSingleBatch(endpointName, batch, originalMessagesById))
442+
.forEach(futures::add);
431443
if (!smallGroups.isEmpty()) {
432-
smallGroups.sort((a, b) -> Integer.compare(b.size(), a.size()));
433-
List<List<Message>> packedBatches = new ArrayList<>();
434-
for (List<Message> group : smallGroups) {
435-
boolean packed = false;
436-
for (List<Message> batch : packedBatches) {
437-
if (batch.size() + group.size() <= SQS_MAX_BATCH_SIZE) {
438-
batch.addAll(group);
439-
packed = true;
440-
break;
441-
}
442-
}
443-
if (!packed) {
444-
packedBatches.add(new ArrayList<>(group));
444+
binPackSmallFifoGroups(smallGroups, SQS_MAX_BATCH_SIZE).stream()
445+
.map(batch -> sendSingleBatch(endpointName, batch, originalMessagesById)).forEach(futures::add);
446+
}
447+
return combineBatchFutures(futures);
448+
}
449+
450+
/**
451+
* Bin-pack small FIFO groups into shared batches using first-fit decreasing algorithm. Each group is kept whole
452+
* within a single batch. Groups are sorted by size descending before packing to minimize the number of batches.
453+
* @param smallGroups groups with size < maxBatchSize
454+
* @param maxBatchSize the maximum number of messages per batch (SQS limit is 10)
455+
* @return packed batches, each containing one or more whole groups
456+
*/
457+
protected static List<List<Message>> binPackSmallFifoGroups(List<List<Message>> smallGroups, int maxBatchSize) {
458+
smallGroups.sort((a, b) -> Integer.compare(b.size(), a.size()));
459+
List<List<Message>> packedBatches = new ArrayList<>();
460+
for (List<Message> group : smallGroups) {
461+
boolean packed = false;
462+
for (List<Message> batch : packedBatches) {
463+
if (batch.size() + group.size() <= maxBatchSize) {
464+
batch.addAll(group);
465+
packed = true;
466+
break;
445467
}
446468
}
447-
packedBatches.stream().map(batch -> sendSingleBatch(endpointName, batch, originalMessagesById))
448-
.forEach(futures::add);
469+
if (!packed) {
470+
packedBatches.add(new ArrayList<>(group));
471+
}
449472
}
450-
return combineBatchFutures(futures);
473+
return packedBatches;
451474
}
452475

453476
private <T> CompletableFuture<SendResult.Batch<T>> combineBatchFutures(

‎spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateTests.java‎

Lines changed: 18 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import java.util.concurrent.CompletionException;
4242
import java.util.concurrent.atomic.AtomicInteger;
4343
import java.util.function.Consumer;
44+
import java.util.stream.Collectors;
4445
import java.util.stream.IntStream;
4546
import org.junit.jupiter.api.BeforeEach;
4647
import org.junit.jupiter.api.Test;
@@ -681,104 +682,32 @@ void shouldGroupMessagesByMessageGroupIdForFifoQueues() {
681682

682683
@Test
683684
void shouldBinPackSmallFifoGroupsIntoSharedBatches() {
684-
String queue = "test-queue.fifo";
685-
String groupA = "group-a";
686-
String groupB = "group-b";
687-
String groupC = "group-c";
688-
String groupD = "group-d";
689-
List<Message<String>> messages = new ArrayList<>();
690-
messages.addAll(createMessagesForGroup(groupA, 3));
691-
messages.addAll(createMessagesForGroup(groupB, 3));
692-
messages.addAll(createMessagesForGroup(groupC, 3));
693-
messages.addAll(createMessagesForGroup(groupD, 3));
694-
695-
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
696-
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
697-
.willReturn(CompletableFuture.completedFuture(urlResponse));
698-
mockQueueAttributes(mockClient, Map.of());
699-
700-
List<SendMessageBatchRequest> captured = new ArrayList<>();
701-
given(mockClient.sendMessageBatch(any(SendMessageBatchRequest.class))).willAnswer(invocation -> {
702-
SendMessageBatchRequest request = invocation.getArgument(0);
703-
captured.add(request);
704-
return CompletableFuture.completedFuture(
705-
SendMessageBatchResponse.builder().successful(successEntries(request.entries())).build());
706-
});
707-
708-
SqsOperations template = SqsTemplate.newSyncTemplate(mockClient);
709-
SendResult.Batch<String> result = template.sendMany(queue, messages);
710-
711-
assertThat(result.successful()).hasSize(12);
712-
assertThat(result.failed()).isEmpty();
713-
assertThat(captured).hasSize(2);
714-
assertThat(captured.get(0).entries()).hasSize(9);
715-
assertThat(captured.get(1).entries()).hasSize(3);
685+
var groups = new ArrayList<>(List.of(sqsMessages(3), sqsMessages(3), sqsMessages(3), sqsMessages(3)));
686+
var packed = SqsTemplate.binPackSmallFifoGroups(groups, 10);
687+
assertThat(packed).hasSize(2);
688+
assertThat(packed.get(0)).hasSize(9);
689+
assertThat(packed.get(1)).hasSize(3);
716690
}
717691

718692
@Test
719-
void shouldBinPackFifoGroupsOfDifferentSizes() {
720-
String queue = "test-queue.fifo";
721-
List<Message<String>> messages = new ArrayList<>();
722-
messages.addAll(createMessagesForGroup("group-4", 4));
723-
messages.addAll(createMessagesForGroup("group-6", 6));
724-
725-
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
726-
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
727-
.willReturn(CompletableFuture.completedFuture(urlResponse));
728-
mockQueueAttributes(mockClient, Map.of());
729-
730-
List<SendMessageBatchRequest> captured = new ArrayList<>();
731-
given(mockClient.sendMessageBatch(any(SendMessageBatchRequest.class))).willAnswer(invocation -> {
732-
SendMessageBatchRequest request = invocation.getArgument(0);
733-
captured.add(request);
734-
return CompletableFuture.completedFuture(
735-
SendMessageBatchResponse.builder().successful(successEntries(request.entries())).build());
736-
});
737-
738-
SqsOperations template = SqsTemplate.newSyncTemplate(mockClient);
739-
SendResult.Batch<String> result = template.sendMany(queue, messages);
740-
741-
assertThat(result.successful()).hasSize(10);
742-
assertThat(result.failed()).isEmpty();
743-
assertThat(captured).hasSize(1);
744-
assertThat(captured.get(0).entries()).hasSize(10);
693+
void shouldPackGroupsThatFitExactly() {
694+
var groups = new ArrayList<>(List.of(sqsMessages(4), sqsMessages(6)));
695+
var packed = SqsTemplate.binPackSmallFifoGroups(groups, 10);
696+
assertThat(packed).hasSize(1);
697+
assertThat(packed.get(0)).hasSize(10);
745698
}
746699

747700
@Test
748-
void shouldBinPackLargeAndSmallGroupsTogether() {
749-
String queue = "test-queue.fifo";
750-
List<Message<String>> messages = new ArrayList<>();
751-
messages.addAll(createMessagesForGroup("large-a", 15));
752-
messages.addAll(createMessagesForGroup("large-b", 12));
753-
messages.addAll(createMessagesForGroup("small-a", 3));
754-
messages.addAll(createMessagesForGroup("small-b", 2));
755-
756-
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
757-
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
758-
.willReturn(CompletableFuture.completedFuture(urlResponse));
759-
mockQueueAttributes(mockClient, Map.of());
760-
761-
List<SendMessageBatchRequest> captured = new ArrayList<>();
762-
given(mockClient.sendMessageBatch(any(SendMessageBatchRequest.class))).willAnswer(invocation -> {
763-
SendMessageBatchRequest request = invocation.getArgument(0);
764-
captured.add(request);
765-
return CompletableFuture.completedFuture(
766-
SendMessageBatchResponse.builder().successful(successEntries(request.entries())).build());
767-
});
768-
769-
SqsOperations template = SqsTemplate.newSyncTemplate(mockClient);
770-
SendResult.Batch<String> result = template.sendMany(queue, messages);
771-
772-
assertThat(result.successful()).hasSize(32);
773-
assertThat(result.failed()).isEmpty();
774-
assertThat(captured).hasSize(5);
701+
void shouldSeparateSmallGroupsWhenBatchIsFull() {
702+
var groups = new ArrayList<>(List.of(sqsMessages(6), sqsMessages(5), sqsMessages(5)));
703+
var packed = SqsTemplate.binPackSmallFifoGroups(groups, 10);
704+
assertThat(packed).hasSize(2);
775705
}
776706

777-
private static List<Message<String>> createMessagesForGroup(String groupId, int count) {
707+
private static List<software.amazon.awssdk.services.sqs.model.Message> sqsMessages(int count) {
778708
return IntStream.range(0, count)
779-
.mapToObj(i -> MessageBuilder.withPayload("payload-" + groupId + "-" + i)
780-
.setHeader(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, groupId).build())
781-
.toList();
709+
.mapToObj(i -> software.amazon.awssdk.services.sqs.model.Message.builder().build())
710+
.collect(Collectors.toCollection(ArrayList::new));
782711
}
783712

784713
@Test

0 commit comments

Comments
 (0)