Skip to content

Commit aaba5fc

Browse files
joseiedojoseiedo
committed
update with upstream main
2 parents c273aae + a123519 commit aaba5fc

6 files changed

Lines changed: 89 additions & 36 deletions

File tree

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
import java.util.ArrayList;
1919
import java.util.Collection;
2020
import java.util.List;
21-
import java.util.stream.Collectors;
2221
import java.util.stream.IntStream;
2322

2423
public class CollectionUtils {
@@ -28,7 +27,7 @@ public static <T> List<Collection<T>> partition(Collection<T> messagesToAck, int
2827
int totalSize = messagesToUse.size();
2928
return IntStream.rangeClosed(0, (totalSize - 1) / pageSize)
3029
.mapToObj(index -> messagesToUse.subList(index * pageSize, Math.min((index + 1) * pageSize, totalSize)))
31-
.collect(Collectors.toList());
30+
.toList();
3231
}
3332

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

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ public static <T> T getHeader(Message<?> message, String headerName, Class<T> cl
8181
*/
8282
public static <T, U> Collection<T> getHeader(Collection<Message<U>> messages, String headerName,
8383
Class<T> classToCast) {
84-
return messages.stream().map(msg -> getHeader(msg, headerName, classToCast)).collect(Collectors.toList());
84+
return messages.stream().map(msg -> getHeader(msg, headerName, classToCast)).toList();
8585
}
8686

8787
/**

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
import io.awspring.cloud.sqs.MessageHeaderUtils;
1919
import java.util.Collection;
2020
import java.util.concurrent.CompletableFuture;
21-
import java.util.stream.Collectors;
2221
import org.slf4j.Logger;
2322
import org.slf4j.LoggerFactory;
2423
import org.springframework.messaging.Message;
@@ -61,7 +60,7 @@ public QueueMessageVisibility(SqsAsyncClient amazonSqsAsync, String queueUrl, St
6160
public BatchVisibility toBatchVisibility(Collection<Message<?>> messages) {
6261
return new QueueMessageBatchVisibility(this.sqsAsyncClient, this.queueUrl, messages.stream().map(
6362
message -> MessageHeaderUtils.getHeader(message, SqsHeaders.SQS_RECEIPT_HANDLE_HEADER, String.class))
64-
.collect(Collectors.toList()));
63+
.toList());
6564
}
6665

6766
@Override

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import java.util.Collection;
2222
import java.util.Collections;
2323
import java.util.List;
24-
import java.util.stream.Collectors;
2524
import org.jspecify.annotations.Nullable;
2625
import org.springframework.util.Assert;
2726
import software.amazon.awssdk.services.sqs.model.MessageSystemAttributeName;
@@ -211,7 +210,7 @@ public SqsContainerOptionsBuilder messageSystemAttributeNames(
211210
Collection<MessageSystemAttributeName> messageSystemAttributeNames) {
212211
Assert.notEmpty(messageSystemAttributeNames, "messageSystemAttributeNames cannot be empty");
213212
this.messageSystemAttributeNames = messageSystemAttributeNames.stream()
214-
.map(MessageSystemAttributeName::toString).collect(Collectors.toList());
213+
.map(MessageSystemAttributeName::toString).toList();
215214
return this;
216215
}
217216

‎spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java‎

Lines changed: 51 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import java.util.ArrayList;
2626
import java.util.Collection;
2727
import java.util.Collections;
28+
import java.util.HashSet;
2829
import java.util.List;
2930
import java.util.Set;
3031
import java.util.concurrent.CompletableFuture;
@@ -98,14 +99,15 @@ private CompletableFuture<Void> deleteMessages(Collection<Message<T>> messagesTo
9899
MessageHeaderUtils.getId(messagesToAck));
99100
StopWatch watch = new StopWatch();
100101
watch.start();
102+
List<Message<T>> orderedMessages = new ArrayList<>(messagesToAck);
101103
return CompletableFutures.exceptionallyCompose(this.sqsAsyncClient
102-
.deleteMessageBatch(createDeleteMessageBatchRequest(messagesToAck)).thenCompose(
103-
response -> handleDeleteMessageBatchResponse(messagesToAck, response)),
104-
t -> toAcknowledgementFailure(messagesToAck, t))
105-
.whenComplete((v, t) -> logAckResult(messagesToAck, t, watch));
104+
.deleteMessageBatch(createDeleteMessageBatchRequest(orderedMessages)).thenCompose(
105+
response -> handleDeleteMessageBatchResponse(orderedMessages, response)),
106+
t -> toAcknowledgementFailure(orderedMessages, t))
107+
.whenComplete((v, t) -> logAckResult(orderedMessages, t, watch));
106108
}
107109

108-
private CompletableFuture<Void> handleDeleteMessageBatchResponse(Collection<Message<T>> messagesToAck,
110+
private CompletableFuture<Void> handleDeleteMessageBatchResponse(List<Message<T>> messagesToAck,
109111
DeleteMessageBatchResponse response) {
110112
if (!response.failed().isEmpty()) {
111113
return CompletableFutures.<Void>failedFuture(createPartialFailureException(messagesToAck, response));
@@ -122,51 +124,73 @@ private CompletableFuture<Void> toAcknowledgementFailure(Collection<Message<T>>
122124
return CompletableFutures.<Void>failedFuture(createAcknowledgementException(messagesToAck, cause));
123125
}
124126

125-
private SqsAcknowledgementException createPartialFailureException(Collection<Message<T>> messages,
127+
private SqsAcknowledgementException createPartialFailureException(List<Message<T>> messages,
126128
DeleteMessageBatchResponse response) {
127-
Set<String> messageIds = messages.stream().map(MessageHeaderUtils::getId).collect(Collectors.toSet());
128-
Set<String> failedIds = response.failed().stream()
129-
.map(BatchResultErrorEntry::id)
130-
.collect(Collectors.toSet());
129+
Set<Integer> failedIndices = new HashSet<>();
130+
boolean allIdsCorrelated = true;
131+
for (BatchResultErrorEntry errorEntry : response.failed()) {
132+
Integer index = parseBatchEntryIndex(errorEntry.id());
133+
if (index == null || index < 0 || index >= messages.size()) {
134+
allIdsCorrelated = false;
135+
break;
136+
}
137+
failedIndices.add(index);
138+
}
131139

132-
if (!messageIds.containsAll(failedIds)) {
140+
if (!allIdsCorrelated) {
141+
Set<String> rawFailedIds = response.failed().stream()
142+
.map(BatchResultErrorEntry::id)
143+
.collect(Collectors.toSet());
133144
logger.warn("Could not correlate all acknowledgement failure ids in queue {}: {}", this.queueName,
134-
failedIds);
135-
return new SqsAcknowledgementException("Could not correlate acknowledgement failure ids: " + failedIds,
145+
rawFailedIds);
146+
return new SqsAcknowledgementException("Could not correlate acknowledgement failure ids: " + rawFailedIds,
136147
Collections.emptyList(), messages.stream().map(msg -> (Message<?>) msg).collect(Collectors.toList()),
137148
this.queueUrl, null);
138149
}
139150

140151
List<Message<?>> successfulMessages = new ArrayList<>();
141152
List<Message<?>> failedMessages = new ArrayList<>();
142153

143-
for(Message<T> msg : messages) {
144-
if(failedIds.contains(MessageHeaderUtils.getId(msg))) {
145-
failedMessages.add(msg);
154+
for (int i = 0; i < messages.size(); i++) {
155+
if (failedIndices.contains(i)) {
156+
failedMessages.add(messages.get(i));
146157
} else {
147-
successfulMessages.add(msg);
158+
successfulMessages.add(messages.get(i));
148159
}
149160
}
150161

151-
logger.warn("Some messages could not be acknowledged in queue {}: {}", this.queueName, failedIds);
162+
Set<String> failedMessageIds = failedMessages.stream()
163+
.map(MessageHeaderUtils::getId)
164+
.collect(Collectors.toSet());
165+
logger.warn("Some messages could not be acknowledged in queue {}: {}", this.queueName, failedMessageIds);
152166

153-
return new SqsAcknowledgementException("Error acknowledging messages " + failedIds, successfulMessages,
167+
return new SqsAcknowledgementException("Error acknowledging messages " + failedMessageIds, successfulMessages,
154168
failedMessages, this.queueUrl, null);
155169
}
156170

157-
private DeleteMessageBatchRequest createDeleteMessageBatchRequest(Collection<Message<T>> messagesToAck) {
158-
return DeleteMessageBatchRequest
159-
.builder()
160-
.queueUrl(this.queueUrl)
161-
.entries(messagesToAck.stream().map(this::toDeleteMessageEntry).collect(Collectors.toList()))
162-
.build();
171+
private Integer parseBatchEntryIndex(String id) {
172+
try {
173+
return Integer.valueOf(id);
174+
}
175+
catch (NumberFormatException ex) {
176+
return null;
177+
}
178+
}
179+
180+
private DeleteMessageBatchRequest createDeleteMessageBatchRequest(List<Message<T>> messagesToAck) {
181+
List<DeleteMessageBatchRequestEntry> entries = new ArrayList<>(messagesToAck.size());
182+
for (int i = 0; i < messagesToAck.size(); i++) {
183+
entries.add(toDeleteMessageEntry(messagesToAck.get(i), i));
184+
}
185+
return DeleteMessageBatchRequest.builder().queueUrl(this.queueUrl).entries(entries).build();
163186
}
164187

165-
private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message<T> message) {
188+
// Positional index keeps the batch-local id unique even when SQS redelivers the same message id.
189+
private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message<T> message, int index) {
166190
return DeleteMessageBatchRequestEntry
167191
.builder()
168192
.receiptHandle(MessageHeaderUtils.getHeaderAsString(message, SqsHeaders.SQS_RECEIPT_HANDLE_HEADER))
169-
.id(MessageHeaderUtils.getId(message))
193+
.id(Integer.toString(index))
170194
.build();
171195
}
172196
// @formatter:on

‎spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,12 @@
2626
import io.awspring.cloud.sqs.SqsAcknowledgementException;
2727
import io.awspring.cloud.sqs.listener.QueueAttributes;
2828
import io.awspring.cloud.sqs.listener.SqsHeaders;
29+
import io.awspring.cloud.sqs.support.converter.MessagingMessageHeaders;
2930
import java.util.Collection;
3031
import java.util.Collections;
3132
import java.util.List;
33+
import java.util.Map;
34+
import java.util.UUID;
3235
import java.util.concurrent.CompletableFuture;
3336
import java.util.concurrent.CompletionException;
3437
import org.junit.jupiter.api.Test;
@@ -151,8 +154,8 @@ void shouldWrapPartialBatchFailure() {
151154
given(queueAttributes.getQueueName()).willReturn(queueName);
152155
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);
153156

154-
BatchResultErrorEntry failedEntry = BatchResultErrorEntry.builder().id(failedMessageHeaders.getId().toString())
155-
.code("ReceiptHandleIsInvalid").message("Receipt handle expired").build();
157+
BatchResultErrorEntry failedEntry = BatchResultErrorEntry.builder().id("0").code("ReceiptHandleIsInvalid")
158+
.message("Receipt handle expired").build();
156159

157160
DeleteMessageBatchResponse partialFailureResponse = DeleteMessageBatchResponse.builder().failed(failedEntry)
158161
.build();
@@ -172,6 +175,35 @@ void shouldWrapPartialBatchFailure() {
172175
});
173176
}
174177

178+
@Test
179+
void shouldUseUniqueBatchEntryIdsWhenMessageIdIsDuplicated() throws Exception {
180+
UUID sharedMessageId = UUID.randomUUID();
181+
MessageHeaders firstHeaders = new MessagingMessageHeaders(
182+
Map.of(SqsHeaders.SQS_RECEIPT_HANDLE_HEADER, receiptHandle), sharedMessageId);
183+
MessageHeaders secondHeaders = new MessagingMessageHeaders(
184+
Map.of(SqsHeaders.SQS_RECEIPT_HANDLE_HEADER, secondReceiptHandle), sharedMessageId);
185+
Collection<Message<String>> messagesToAck = List.of(message, secondMessage);
186+
given(message.getHeaders()).willReturn(firstHeaders);
187+
given(secondMessage.getHeaders()).willReturn(secondHeaders);
188+
given(queueAttributes.getQueueName()).willReturn(queueName);
189+
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);
190+
given(sqsAsyncClient.deleteMessageBatch(any(DeleteMessageBatchRequest.class)))
191+
.willReturn(CompletableFuture.completedFuture(DeleteMessageBatchResponse.builder().build()));
192+
193+
SqsAcknowledgementExecutor<String> executor = new SqsAcknowledgementExecutor<>();
194+
executor.setSqsAsyncClient(sqsAsyncClient);
195+
executor.setQueueAttributes(queueAttributes);
196+
executor.execute(messagesToAck).get();
197+
198+
ArgumentCaptor<DeleteMessageBatchRequest> requestCaptor = ArgumentCaptor
199+
.forClass(DeleteMessageBatchRequest.class);
200+
verify(sqsAsyncClient).deleteMessageBatch(requestCaptor.capture());
201+
List<DeleteMessageBatchRequestEntry> entries = requestCaptor.getValue().entries();
202+
assertThat(entries).extracting(DeleteMessageBatchRequestEntry::id).doesNotHaveDuplicates();
203+
assertThat(entries).extracting(DeleteMessageBatchRequestEntry::receiptHandle).containsExactly(receiptHandle,
204+
secondReceiptHandle);
205+
}
206+
175207
@Test
176208
void shouldTreatAllMessagesAsFailedIfAwsFailureIdCannotBeCorrelated() {
177209
Collection<Message<String>> messagesToAck = List.of(message, secondMessage);

0 commit comments

Comments
 (0)