Skip to content

Fix/enhance sqs partial acknowledgement handling #1562

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

Merged
tomazfernandes merged 13 commits into from
Mar 8, 2026
+152 −7
Merged

Fix/enhance sqs partial acknowledgement handling #1562

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
c006b95
Fix : Add looging for SQS Partial Acknowledgement
co2plant Jan 25, 2026
fde00e3
Style : Modify code style to suit the characteristics of the existing…
co2plant Jan 25, 2026
e12bd5f
Fix : Test code changes due to added logging
co2plant Jan 25, 2026
de0e709
Test : Add Test for partialBatchFailure
co2plant Jan 25, 2026
78d3b26
Fix : Added logging of failed ID list
co2plant Jan 25, 2026
b278109
Fix: Added success/failure lists to SqsAcknowledgementException
co2plant Jan 25, 2026
1730ba5
Refactoring: Extract the createPartialFailureException helper method
co2plant Jan 25, 2026
4bdcd04
Refactoring: Separate the batch response processing deletion part fro…
co2plant Feb 16, 2026
fc08919
Fix: improve partial SQS acknowledgement failure handling
co2plant Feb 18, 2026
e3fa4a7
Docs: document partial acknowledgement failure handling in callbacks
co2plant Feb 18, 2026
38e50ea
Fix: add fail-safe handling for uncorrelated SQS acknowledgement fail…
co2plant Feb 22, 2026
932f090
Test: add coverage for uncorrelated acknowledgement failure ids
co2plant Feb 22, 2026
7b71da1
Docs: document fail-safe correlation behavior in acknowledgement resu…
co2plant Feb 22, 2026
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
45 changes: 41 additions & 4 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 @@ -22,9 +22,11 @@
import io.awspring.cloud.sqs.listener.QueueAttributesAware;
import io.awspring.cloud.sqs.listener.SqsAsyncClientAware;
import io.awspring.cloud.sqs.listener.SqsHeaders;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.stream.Collectors;
Expand All @@ -34,8 +36,10 @@
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse;

/**
* {@link AcknowledgementExecutor} implementation for SQS queues. Handle the messages deletion, usually requested by an
Expand Down Expand Up @@ -96,11 +100,44 @@ private CompletableFuture<Void> deleteMessages(Collection<Message<T>> messagesTo
watch.start();
return CompletableFutures.exceptionallyCompose(this.sqsAsyncClient
.deleteMessageBatch(createDeleteMessageBatchRequest(messagesToAck))
.thenRun(() -> {}),
t -> CompletableFutures.failedFuture(createAcknowledgementException(messagesToAck, t)))
.thenCompose(response -> {
if (!response.failed().isEmpty()) {
Comment thread
co2plant marked this conversation as resolved.
Outdated
return CompletableFutures.<Void> failedFuture(
createPartialFailureException(messagesToAck, response));
}
return CompletableFuture.<Void>completedFuture(null);
}),
t -> CompletableFutures.<Void>failedFuture(createAcknowledgementException(messagesToAck, t)))
.whenComplete((v, t) -> logAckResult(messagesToAck, t, watch));
}

private SqsAcknowledgementException createPartialFailureException(Collection<Message<T>> messages, DeleteMessageBatchResponse response){
Set<String> failedIds = response.failed().stream()
.map(BatchResultErrorEntry::id)
.collect(Collectors.toSet());

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

for(Message<T> msg : messages) {
if(failedIds.contains(MessageHeaderUtils.getId(msg))) {
failedMessages.add(msg);
} else {
successfulMessages.add(msg);
}
}

logger.warn("Some messages could not be acknowledged in queue {}: {}",
this.queueName, failedIds);

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

private DeleteMessageBatchRequest createDeleteMessageBatchRequest(Collection<Message<T>> messagesToAck) {
return DeleteMessageBatchRequest
.builder()
Expand All @@ -113,7 +150,7 @@ private DeleteMessageBatchRequestEntry toDeleteMessageEntry(Message<T> message)
return DeleteMessageBatchRequestEntry
.builder()
.receiptHandle(MessageHeaderUtils.getHeaderAsString(message, SqsHeaders.SQS_RECEIPT_HANDLE_HEADER))
.id(UUID.randomUUID().toString())
.id(MessageHeaderUtils.getId(message))
.build();
}
// @formatter:on
Expand Down
28 changes: 27 additions & 1 deletion .../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 @@ -38,8 +38,10 @@
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse;

/**
* Tests for {@link SqsAcknowledgementExecutor}.
Expand Down Expand Up @@ -74,7 +76,7 @@ void shouldDeleteMessages() throws Exception {
given(queueAttributes.getQueueName()).willReturn(queueName);
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);
given(sqsAsyncClient.deleteMessageBatch(any(DeleteMessageBatchRequest.class)))
.willReturn(CompletableFuture.completedFuture(null));
.willReturn(CompletableFuture.completedFuture(DeleteMessageBatchResponse.builder().build()));

SqsAcknowledgementExecutor<String> executor = new SqsAcknowledgementExecutor<>();
executor.setSqsAsyncClient(sqsAsyncClient);
Expand Down Expand Up @@ -127,4 +129,28 @@ void shouldWrapIfErrorIsThrown() {
.extracting(SqsAcknowledgementException::getQueue).isEqualTo(queueUrl);
}

@Test
void shouldWrapPartialBatchFailure() {
Collection<Message<String>> messages = Collections.singletonList(message);
given(message.getHeaders()).willReturn(messageHeaders);
given(queueAttributes.getQueueName()).willReturn(queueName);
given(queueAttributes.getQueueUrl()).willReturn(queueUrl);

BatchResultErrorEntry failedEntry = BatchResultErrorEntry.builder().id("test-id").code("ReceiptHandleIsInvalid")
Comment thread
co2plant marked this conversation as resolved.
Outdated
.message("Receipt handle expired").build();

DeleteMessageBatchResponse partialFailureResponse = DeleteMessageBatchResponse.builder().failed(failedEntry)
.build();

given(sqsAsyncClient.deleteMessageBatch(any(DeleteMessageBatchRequest.class)))
.willReturn(CompletableFuture.completedFuture(partialFailureResponse));

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

assertThatThrownBy(() -> executor.execute(messages).join()).isInstanceOf(CompletionException.class)
.hasCauseInstanceOf(SqsAcknowledgementException.class);
Comment thread
co2plant marked this conversation as resolved.
Outdated
}

}
Loading