diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java index 08b4b3604..1e8c62103 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutor.java @@ -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; @@ -98,14 +99,15 @@ private CompletableFuture deleteMessages(Collection> messagesTo MessageHeaderUtils.getId(messagesToAck)); StopWatch watch = new StopWatch(); watch.start(); + List> 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 handleDeleteMessageBatchResponse(Collection> messagesToAck, + private CompletableFuture handleDeleteMessageBatchResponse(List> messagesToAck, DeleteMessageBatchResponse response) { if (!response.failed().isEmpty()) { return CompletableFutures.failedFuture(createPartialFailureException(messagesToAck, response)); @@ -122,17 +124,26 @@ private CompletableFuture toAcknowledgementFailure(Collection> return CompletableFutures.failedFuture(createAcknowledgementException(messagesToAck, cause)); } - private SqsAcknowledgementException createPartialFailureException(Collection> messages, + private SqsAcknowledgementException createPartialFailureException(List> messages, DeleteMessageBatchResponse response) { - Set messageIds = messages.stream().map(MessageHeaderUtils::getId).collect(Collectors.toSet()); - Set failedIds = response.failed().stream() - .map(BatchResultErrorEntry::id) - .collect(Collectors.toSet()); + Set 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 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); } @@ -140,33 +151,46 @@ private SqsAcknowledgementException createPartialFailureException(Collection> successfulMessages = new ArrayList<>(); List> failedMessages = new ArrayList<>(); - for(Message 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 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> 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> messagesToAck) { + List 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 message) { + // Positional index keeps the batch-local id unique even when SQS redelivers the same message id. + private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message 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 diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java index f7828231a..91e4be1ca 100644 --- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java +++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/acknowledgement/SqsAcknowledgementExecutorTests.java @@ -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; @@ -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(); @@ -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> 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 executor = new SqsAcknowledgementExecutor<>(); + executor.setSqsAsyncClient(sqsAsyncClient); + executor.setQueueAttributes(queueAttributes); + executor.execute(messagesToAck).get(); + + ArgumentCaptor requestCaptor = ArgumentCaptor + .forClass(DeleteMessageBatchRequest.class); + verify(sqsAsyncClient).deleteMessageBatch(requestCaptor.capture()); + List entries = requestCaptor.getValue().entries(); + assertThat(entries).extracting(DeleteMessageBatchRequestEntry::id).doesNotHaveDuplicates(); + assertThat(entries).extracting(DeleteMessageBatchRequestEntry::receiptHandle).containsExactly(receiptHandle, + secondReceiptHandle); + } + @Test void shouldTreatAllMessagesAsFailedIfAwsFailureIdCannotBeCorrelated() { Collection> messagesToAck = List.of(message, secondMessage);