Skip to content

Fix duplicate DeleteMessageBatchRequestEntry ids when SQS redelivers a message within one ack batch #1654

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BK202503 wants to merge 1 commit into
base: main
Choose a base branch
from
+85 −29
Open

Fix duplicate DeleteMessageBatchRequestEntry ids when SQS redelivers a message within one ack batch #1654

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 51 additions & 27 deletions .../main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -98,14 +99,15 @@ private CompletableFuture<Void> deleteMessages(Collection<Message<T>> messagesTo
MessageHeaderUtils.getId(messagesToAck));
StopWatch watch = new StopWatch();
watch.start();
List<Message<T>> orderedMessages = new ArrayList<>(messagesToAck);
return CompletableFutures.exceptionallyCompose(this.sqsAsyncClient
.deleteMessageBatch(createDeleteMessageBatchRequest(messagesToAck)).thenCompose(
response -> handleDeleteMessageBatchResponse(messagesToAck, response)),
t -> toAcknowledgementFailure(messagesToAck, t))
.whenComplete((v, t) -> logAckResult(messagesToAck, t, watch));
.deleteMessageBatch(createDeleteMessageBatchRequest(orderedMessages)).thenCompose(
response -> handleDeleteMessageBatchResponse(orderedMessages, response)),
t -> toAcknowledgementFailure(orderedMessages, t))
.whenComplete((v, t) -> logAckResult(orderedMessages, t, watch));
}

private CompletableFuture<Void> handleDeleteMessageBatchResponse(Collection<Message<T>> messagesToAck,
private CompletableFuture<Void> handleDeleteMessageBatchResponse(List<Message<T>> messagesToAck,
DeleteMessageBatchResponse response) {
if (!response.failed().isEmpty()) {
return CompletableFutures.<Void>failedFuture(createPartialFailureException(messagesToAck, response));
Expand All @@ -122,51 +124,73 @@ private CompletableFuture<Void> toAcknowledgementFailure(Collection<Message<T>>
return CompletableFutures.<Void>failedFuture(createAcknowledgementException(messagesToAck, cause));
}

private SqsAcknowledgementException createPartialFailureException(Collection<Message<T>> messages,
private SqsAcknowledgementException createPartialFailureException(List<Message<T>> messages,
DeleteMessageBatchResponse response) {
Set<String> messageIds = messages.stream().map(MessageHeaderUtils::getId).collect(Collectors.toSet());
Set<String> failedIds = response.failed().stream()
.map(BatchResultErrorEntry::id)
.collect(Collectors.toSet());
Set<Integer> failedIndices = new HashSet<>();
boolean allIdsCorrelated = true;
for (BatchResultErrorEntry errorEntry : response.failed()) {
Integer index = parseBatchEntryIndex(errorEntry.id());
if (index == null || index < 0 || index >= messages.size()) {
allIdsCorrelated = false;
break;
}
failedIndices.add(index);
}

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

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

for(Message<T> msg : messages) {
if(failedIds.contains(MessageHeaderUtils.getId(msg))) {
failedMessages.add(msg);
for (int i = 0; i < messages.size(); i++) {
if (failedIndices.contains(i)) {
failedMessages.add(messages.get(i));
} else {
successfulMessages.add(msg);
successfulMessages.add(messages.get(i));
}
}

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

return new SqsAcknowledgementException("Error acknowledging messages " + failedIds, successfulMessages,
return new SqsAcknowledgementException("Error acknowledging messages " + failedMessageIds, successfulMessages,
failedMessages, this.queueUrl, null);
}

private DeleteMessageBatchRequest createDeleteMessageBatchRequest(Collection<Message<T>> messagesToAck) {
return DeleteMessageBatchRequest
.builder()
.queueUrl(this.queueUrl)
.entries(messagesToAck.stream().map(this::toDeleteMessageEntry).collect(Collectors.toList()))
.build();
private Integer parseBatchEntryIndex(String id) {
try {
return Integer.valueOf(id);
}
catch (NumberFormatException ex) {
return null;
}
}

private DeleteMessageBatchRequest createDeleteMessageBatchRequest(List<Message<T>> messagesToAck) {
List<DeleteMessageBatchRequestEntry> entries = new ArrayList<>(messagesToAck.size());
for (int i = 0; i < messagesToAck.size(); i++) {
entries.add(toDeleteMessageEntry(messagesToAck.get(i), i));
}
return DeleteMessageBatchRequest.builder().queueUrl(this.queueUrl).entries(entries).build();
}

private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message<T> message) {
// Positional index keeps the batch-local id unique even when SQS redelivers the same message id.
private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message<T> message, int index) {
return DeleteMessageBatchRequestEntry
.builder()
.receiptHandle(MessageHeaderUtils.getHeaderAsString(message, SqsHeaders.SQS_RECEIPT_HANDLE_HEADER))
.id(MessageHeaderUtils.getId(message))
.id(Integer.toString(index))
.build();
}
// @formatter:on
Expand Down
36 changes: 34 additions & 2 deletions .../java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,12 @@
import io.awspring.cloud.sqs.SqsAcknowledgementException;
import io.awspring.cloud.sqs.listener.QueueAttributes;
import io.awspring.cloud.sqs.listener.SqsHeaders;
import io.awspring.cloud.sqs.support.converter.MessagingMessageHeaders;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -151,8 +154,8 @@ void shouldWrapPartialBatchFailure() {
given(queueAttributes.getQueueName()).willReturn(queueName);
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);

BatchResultErrorEntry failedEntry = BatchResultErrorEntry.builder().id(failedMessageHeaders.getId().toString())
.code("ReceiptHandleIsInvalid").message("Receipt handle expired").build();
BatchResultErrorEntry failedEntry = BatchResultErrorEntry.builder().id("0").code("ReceiptHandleIsInvalid")
.message("Receipt handle expired").build();

DeleteMessageBatchResponse partialFailureResponse = DeleteMessageBatchResponse.builder().failed(failedEntry)
.build();
Expand All @@ -172,6 +175,35 @@ void shouldWrapPartialBatchFailure() {
});
}

@Test
void shouldUseUniqueBatchEntryIdsWhenMessageIdIsDuplicated() throws Exception {
UUID sharedMessageId = UUID.randomUUID();
MessageHeaders firstHeaders = new MessagingMessageHeaders(
Map.of(SqsHeaders.SQS_RECEIPT_HANDLE_HEADER, receiptHandle), sharedMessageId);
MessageHeaders secondHeaders = new MessagingMessageHeaders(
Map.of(SqsHeaders.SQS_RECEIPT_HANDLE_HEADER, secondReceiptHandle), sharedMessageId);
Collection<Message<String>> messagesToAck = List.of(message, secondMessage);
given(message.getHeaders()).willReturn(firstHeaders);
given(secondMessage.getHeaders()).willReturn(secondHeaders);
given(queueAttributes.getQueueName()).willReturn(queueName);
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);
given(sqsAsyncClient.deleteMessageBatch(any(DeleteMessageBatchRequest.class)))
.willReturn(CompletableFuture.completedFuture(DeleteMessageBatchResponse.builder().build()));

SqsAcknowledgementExecutor<String> executor = new SqsAcknowledgementExecutor<>();
executor.setSqsAsyncClient(sqsAsyncClient);
executor.setQueueAttributes(queueAttributes);
executor.execute(messagesToAck).get();

ArgumentCaptor<DeleteMessageBatchRequest> requestCaptor = ArgumentCaptor
.forClass(DeleteMessageBatchRequest.class);
verify(sqsAsyncClient).deleteMessageBatch(requestCaptor.capture());
List<DeleteMessageBatchRequestEntry> entries = requestCaptor.getValue().entries();
assertThat(entries).extracting(DeleteMessageBatchRequestEntry::id).doesNotHaveDuplicates();
assertThat(entries).extracting(DeleteMessageBatchRequestEntry::receiptHandle).containsExactly(receiptHandle,
secondReceiptHandle);
}

@Test
void shouldTreatAllMessagesAsFailedIfAwsFailureIdCannotBeCorrelated() {
Collection<Message<String>> messagesToAck = List.of(message, secondMessage);
Expand Down