1. Overview
Acknowledgment of messages is a standard mechanism within messaging systems, signaling to the message broker that the message has been received and should not be delivered again. In Amazon’s SQS (Simple Queue Service), acknowledgment is executed through the deletion of messages in the queue.
In this tutorial, we’ll explore the three acknowledgment modes Spring Cloud AWS SQS v3 provides out-of-the-box: ON_SUCCESS, MANUAL, and ALWAYS.
We’ll use an event-driven scenario to illustrate our use cases, leveraging the environment and test setup from the Spring Cloud AWS SQS V3 introductory article.
2. Dependencies
We’ll first import the Spring Cloud AWS Bill of Materials to ensure all dependencies in our pom.xml are compatible with each other:
<dependencyManagement>
<dependencies>
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-aws</artifactId>
<version>3.1.0</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
We’ll also add the Core and SQS starter dependencies:
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-aws-starter-sqs</artifactId>
</dependency>
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-aws-starter</artifactId>
</dependency>
Finally, we’ll add the dependencies required for our tests, namely LocalStack and TestContainers with JUnit 5, the awaitility library for verifying asynchronous message consumption, and AssertJ to handle the assertions:
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>localstack</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
3. Setting up the Local Test Environment
First, we’ll configure a LocalStack environment with Testcontainers for local testing:
@Testcontainers
public class BaseSqsLiveTest {
private static final String LOCAL_STACK_VERSION = "localstack/localstack:2.3.2";
@Container
static LocalStackContainer localStack = new LocalStackContainer(DockerImageName.parse(LOCAL_STACK_VERSION));
@DynamicPropertySource
static void overrideProperties(DynamicPropertyRegistry registry) {
registry.add("spring.cloud.aws.region.static", () -> localStack.getRegion());
registry.add("spring.cloud.aws.credentials.access-key", () -> localStack.getAccessKey());
registry.add("spring.cloud.aws.credentials.secret-key", () -> localStack.getSecretKey());
registry.add("spring.cloud.aws.sqs.endpoint", () -> localStack.getEndpointOverride(SQS)
.toString());
}
}
Although this setup makes testing easy and repeatable, note that the code in this tutorial can also be used to target AWS directly.
4. Setting up the Queue Names
By default, Spring Cloud AWS SQS automatically creates the queues specified in any @SqsListener annotated method. As a first setup step, we’ll define queue names in our application.yaml file:
events:
queues:
order-processing-retry-queue: order_processing_retry_queue
order-processing-async-queue: order_processing_async_queue
order-processing-no-retries-queue: order_processing_no_retries_queue
acknowledgment:
order-processing-no-retries-queue: ALWAYS
The acknowledgment property ALWAYS will also be used by one of our listeners.
Let’s also add a few productIds to the same file to use throughout our examples:
product:
id:
smartphone: 123e4567-e89b-12d3-a456-426614174000
wireless-headphones: 123e4567-e89b-12d3-a456-426614174001
laptop: 123e4567-e89b-12d3-a456-426614174002
tablet: 123e4567-e89b-12d3-a456-426614174004
To get these properties as a POJO in our application, we’ll create two @ConfigurationProperties classes, one for the queues:
@ConfigurationProperties(prefix = "events.queues")
public class EventsQueuesProperties {
private String orderProcessingRetryQueue;
private String orderProcessingAsyncQueue;
private String orderProcessingNoRetriesQueue;
// getters and setters
}
And another for the products:
@ConfigurationProperties("product.id")
public class ProductIdProperties {
private UUID smartphone;
private UUID wirelessHeadphones;
private UUID laptop;
// getters and setters
}
Lastly, we enable the configuration properties in a @Configuration class using @EnableConfigurationProperties:
@EnableConfigurationProperties({ EventsQueuesProperties.class, ProductIdProperties.class})
@Configuration
public class OrderProcessingConfiguration {
}
5. Acknowledgement on Successful Processing
The default acknowledgment mode for @SqsListeners is ON_SUCCESS. In this mode, if the listener method completes its execution without throwing an error, the message gets acknowledged.
To illustrate this behavior, we’ll create a simple listener that will receive an OrderCreatedEvent, check an InventoryService, and, if the requested item and quantity are in stock, change the order status to PROCESSED.
5.1. Creating the Services
Let’s begin by creating our OrderService, which will be responsible for updating the order status:
@Service
public class OrderService {
Map<UUID, OrderStatus> ORDER_STATUS_STORAGE = new ConcurrentHashMap<>();
public void updateOrderStatus(UUID orderId, OrderStatus status) {
ORDER_STATUS_STORAGE.put(orderId, status);
}
public OrderStatus getOrderStatus(UUID orderId) {
return ORDER_STATUS_STORAGE.getOrDefault(orderId, OrderStatus.UNKNOWN);
}
}
Then, we’ll create the InventoryService. We’ll simulate storage using a Map, populating it using ProductIdProperties, which is autowired with values from our application.yaml file:
@Service
public class InventoryService implements InitializingBean {
private ProductIdProperties productIdProperties;
private Map<UUID, Integer> inventory;
public InventoryService(ProductIdProperties productIdProperties) {
this.productIdProperties = productIdProperties;
}
@Override
public void afterPropertiesSet() {
this.inventory = new ConcurrentHashMap<>(Map.of(productIdProperties.getSmartphone(), 10,
productIdProperties.getWirelessHeadphones(), 15,
productIdProperties.getLaptop(), 5);
}
}
The InitializingBean interface provides afterPropertiesSet, which is a lifecycle method that Spring invokes after all dependencies for the bean have been resolved — in our case, the ProductIdProperties bean.
Let’s add a checkInventory method, which verifies whether the inventory has the requested quantity of the product. If the product doesn’t exist, it’ll throw a ProductNotFoundException, and if the product exists but not in sufficient quantity, it’ll throw an OutOfStockException. In the second scenario, we’ll also simulate a random replenishment so that upon a few retries, the processing will eventually succeed:
public void checkInventory(UUID productId, int quantity) {
Integer stock = inventory.get(productId);
if (stock < quantity) {
inventory.put(productId, stock + (int) (Math.random() * 5));
throw new OutOfStockException(
"Product with id %s is out of stock. Quantity requested: %s ".formatted(productId, quantity));
};
inventory.put(productId, stock - quantity);
}
5.2. Creating the Listener
We’re all set to create our first listener. We’ll use the @Component annotation and inject the services through Spring’s constructor dependency injection mechanism:
@Component
public class OrderProcessingListeners {
private static final Logger logger = LoggerFactory.getLogger(OrderProcessingListeners.class);
private InventoryService inventoryService;
private OrderService orderService;
public OrderProcessingListeners(InventoryService inventoryService, OrderService orderService) {
this.inventoryService = inventoryService;
this.orderService = orderService;
}
}
Next, let’s write the listener method:
@SqsListener(value = "${events.queues.order-processing-retry-queue}", id = "retry-order-processing-container", messageVisibilitySeconds = "1")
public void stockCheckRetry(OrderCreatedEvent orderCreatedEvent) {
logger.info("Message received: {}", orderCreatedEvent);
orderService.updateOrderStatus(orderCreatedEvent.id(), OrderStatus.PROCESSING);
inventoryService.checkInventory(orderCreatedEvent.productId(), orderCreatedEvent.quantity());
orderService.updateOrderStatus(orderCreatedEvent.id(), OrderStatus.PROCESSED);
logger.info("Message processed successfully: {}", orderCreatedEvent);
}
The value property is the queue name autowired via the application.yaml. Since ON_SUCCESS is the default acknowledgment mode, we don’t need to specify it in the annotation.Â
5.3. Setting up the Test Class
To assert the logic is working as expected, let’s create a test class:
@SpringBootTest
class OrderProcessingApplicationLiveTest extends BaseSqsLiveTest {
private static final Logger logger = LoggerFactory.getLogger(OrderProcessingApplicationLiveTest.class);
@Autowired
private EventsQueuesProperties eventsQueuesProperties;
@Autowired
private ProductIdProperties productIdProperties;
@Autowired
private SqsTemplate sqsTemplate;
@Autowired
private OrderService orderService;
@Autowired
private MessageListenerContainerRegistry registry;
}
We’ll also add a method named assertQueueIsEmpty. In it we’ll use the autowired MessageListenerContainerRegistry to fetch the container, then stop the container to make sure it’s not consuming any messages. The registry contains all containers created by the @SqsListener annotation:
private void assertQueueIsEmpty(String queueName, String containerId) {
logger.info("Stopping container {}", containerId);
var container = Objects
.requireNonNull(registry.getContainerById(containerId), () -> "could not find container " + containerId);
container.stop();
// ...
}
After the container stops, we’ll use the SqsTemplate to look for messages in the queue. If acknowledgment has succeeded, no messages should be returned. We’ll also set pollTimeout to a value greater than the visibility timeout so that, if the message hasn’t been deleted, it’ll be delivered again in the specified interval.
Here’s the continuation of the assertQueueIsEmpty method:
// ...
logger.info("Checking for messages in queue {}", queueName);
var message = sqsTemplate.receive(from -> from.queue(queueName)
.pollTimeout(Duration.ofSeconds(5)));
assertThat(message).isEmpty();
logger.info("No messages found in queue {}", queueName);
5.4. Testing
In this first test, we’ll send an OrderCreatedEvent to the queue, containing an Order for a product with a quantity greater than what’s in our inventory. When the exception goes through the listener method, it’ll signal to the framework that message processing failed and the message should be delivered again after the message visibility time window has elapsed.
To speed up testing, we set messageVisibilitySeconds to 1 in the annotation, but typically, this configuration is done in the queue itself and defaults to 30 seconds.
We’ll create the event and send it using the auto-configured SqsTemplate that Spring Cloud AWS provides. Then, we’ll use Awaitility to wait for the order status to change to PROCESSED, and, lastly, we’ll assert that the queue is empty, meaning that the acknowledgment succeeded:
@Test
public void givenOnSuccessAcknowledgementMode_whenProcessingThrows_shouldRetry() {
var orderId = UUID.randomUUID();
var queueName = eventsQueuesProperties.getOrderProcessingRetryQueue();
sqsTemplate.send(queueName, new OrderCreatedEvent(orderId, productIdProperties.getLaptop(), 10));
Awaitility.await()
.atMost(Duration.ofMinutes(1))
.until(() -> orderService.getOrderStatus(orderId)
.equals(OrderStatus.PROCESSED));
assertQueueIsEmpty(queueName, "retry-order-processing-container");
}
Notice we’re passing the containerId specified in the @SqsListener annotation to the assertQueueIsEmpty method.
Now we can run the test. First, we’ll ensure Docker is running, and then we’ll execute the test. After the container initialization logs, we should see our application’s log message:
Message received: OrderCreatedEvent[id=83f27bf2-1bd4-460a-9006-d784ec7eff47, productId=123e4567-e89b-12d3-a456-426614174002, quantity=10]
Then, should see one or more failures due to lack of stock:
Caused by: com.baeldung.spring.cloud.aws.sqs.acknowledgement.exception.OutOfStockException: Product with id 123e4567-e89b-12d3-a456-426614174002 is out of stock. Quantity requested: 10
And, since we added the replenishing logic, we should eventually see that the message processing succeeds:
Message processed successfully: OrderCreatedEvent[id=83f27bf2-1bd4-460a-9006-d784ec7eff47, productId=123e4567-e89b-12d3-a456-426614174002, quantity=10]
Lastly, we’ll ensure acknowledgment has succeeded:
INFO 2699 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Stopping container retry-order-processing-container
INFO 2699 --- [main] a.c.s.l.AbstractMessageListenerContainer : Container retry-order-processing-container stopped
INFO 2699 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Checking for messages in queue order_processing_retry_queue
INFO 2699 --- [main] a.s.a.OrderProcessingApplicationLiveTest : No messages found in queue order_processing_retry_queue
Note that “connection refused” errors might be thrown after the test completes — that’s because the Docker container stops before the framework can stop polling for messages. We can safely ignore these errors.
6. Manual Acknowledgement
The framework supports manual acknowledgment of messages, which is useful for scenarios where we need greater control over the acknowledgment process.
6.1. Creating the Listener
To illustrate this, we’ll create an asynchronous scenario where the InventoryService has a slow connection and we want to release the listener thread before it completes:
@SqsListener(value = "${events.queues.order-processing-async-queue}", acknowledgementMode = SqsListenerAcknowledgementMode.MANUAL, id = "async-order-processing-container", messageVisibilitySeconds = "3")
public void slowStockCheckAsynchronous(OrderCreatedEvent orderCreatedEvent, Acknowledgement acknowledgement) {
orderService.updateOrderStatus(orderCreatedEvent.id(), OrderStatus.PROCESSING);
CompletableFuture.runAsync(() -> inventoryService.slowCheckInventory(orderCreatedEvent.productId(), orderCreatedEvent.quantity()))
.thenRun(() -> orderService.updateOrderStatus(orderCreatedEvent.id(), OrderStatus.PROCESSED))
.thenCompose(voidFuture -> acknowledgement.acknowledgeAsync())
.thenRun(() -> logger.info("Message for order {} acknowledged", orderCreatedEvent.id()));
logger.info("Releasing processing thread.");
}
In this logic, we use Java’s CompletableFuture to run the inventory check asynchronously. We added the Acknowledge object to the listener method and SqsListenerAcknowledgementMode.MANUAL to the annotation’s acknowledgementMode property. This property is a String and accepts property placeholders and SpEL. The Acknowledgement object is only available when we set AcknowledgementMode to MANUAL.
Note that, in this example, we leverage Spring Boot auto-configuration, which provides sensible defaults, and the @SqsListener annotation properties to change between acknowledgment modes. An alternative would be declaring a SqsMessageListenerContainerFactory bean, which allows setting up more complex configurations.
6.2. Simulating a Slow Connection
Now, let’s add the slowCheckInventory method to the InventoryService class, simulating a slow connection using Thread.sleep:
public void slowCheckInventory(UUID productId, int quantity) {
simulateBusyConnection();
checkInventory(productId, quantity);
}
private void simulateBusyConnection() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
}
6.3. Testing
Next, let’s write our test:
@Test
public void givenManualAcknowledgementMode_whenManuallyAcknowledge_shouldAcknowledge() {
var orderId = UUID.randomUUID();
var queueName = eventsQueuesProperties.getOrderProcessingAsyncQueue();
sqsTemplate.send(queueName, new OrderCreatedEvent(orderId, productIdProperties.getSmartphone(), 1));
Awaitility.await()
.atMost(Duration.ofMinutes(1))
.until(() -> orderService.getOrderStatus(orderId)
.equals(OrderStatus.PROCESSED));
assertQueueIsEmpty(queueName, "async-order-processing-container");
}
This time, we’re requesting a quantity available in the inventory, so we should see no errors thrown.
When running the test, we’ll see a log message indicating the message was received:
INFO 2786 --- [ing-container-1] c.b.s.c.a.s.a.l.OrderProcessingListeners : Message received: OrderCreatedEvent[id=013740a3-0a45-478a-b085-fbd634fbe66d, productId=123e4567-e89b-12d3-a456-426614174000, quantity=1]
Then, we’ll see the thread release message:
INFO 2786 --- [ing-container-1] c.b.s.c.a.s.a.l.OrderProcessingListeners : Releasing processing thread.
This is because we’re processing and acknowledging the message asynchronously. After about two seconds, we should see the log that the message has been acknowledged:
INFO 2786 --- [onPool-worker-1] c.b.s.c.a.s.a.l.OrderProcessingListeners : Message for order 013740a3-0a45-478a-b085-fbd634fbe66d acknowledged
Finally, we’ll see the logs for stopping the container and asserting that the queue is empty:
INFO 2786 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Stopping container async-order-processing-container
INFO 2786 --- [main] a.c.s.l.AbstractMessageListenerContainer : Container async-order-processing-container stopped
INFO 2786 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Checking for messages in queue order_processing_async_queue
INFO 2786 --- [main] a.s.a.OrderProcessingApplicationLiveTest : No messages found in queue order_processing_async_queue
7. Acknowledgement on Both Success and Error
The last acknowledgment mode we’ll explore is ALWAYS, which causes the framework to acknowledge the message regardless of whether the listener method throws an error.
7.1. Creating the Listener
Let’s simulate a sales event during which our inventory is limited and we don’t want to reprocess any messages, regardless of any failures. We’ll set the acknowledgment mode to ALWAYS using the property we defined earlier in our application.yml:
@SqsListener(value = "${events.queues.order-processing-no-retries-queue}", acknowledgementMode = ${events.acknowledgment.order-processing-no-retries-queue}, id = "no-retries-order-processing-container", messageVisibilitySeconds = "3")
public void stockCheckNoRetries(OrderCreatedEvent orderCreatedEvent) {
logger.info("Message received: {}", orderCreatedEvent);
orderService.updateOrderStatus(orderCreatedEvent.id(), OrderStatus.RECEIVED);
inventoryService.checkInventory(orderCreatedEvent.productId(), orderCreatedEvent.quantity());
logger.info("Message processed: {}", orderCreatedEvent);
}
In the test, we’ll create an order with a quantity greater than our stock:
7.2. Testing
@Test
public void givenAlwaysAcknowledgementMode_whenProcessThrows_shouldAcknowledge() {
var orderId = UUID.randomUUID();
var queueName = eventsQueuesProperties.getOrderProcessingNoRetriesQueue();
sqsTemplate.send(queueName, new OrderCreatedEvent(orderId, productIdProperties.getWirelessHeadphones(), 20));
Awaitility.await()
.atMost(Duration.ofMinutes(1))
.until(() -> orderService.getOrderStatus(orderId)
.equals(OrderStatus.RECEIVED));
assertQueueIsEmpty(queueName, "no-retries-order-processing-container");
}
Now, even when the OutOfStockException is thrown, the message is acknowledged and no retries are attempted for the message:
Message received: OrderCreatedEvent[id=7587f1a2-328f-4791-8559-ee8e85b25259, productId=123e4567-e89b-12d3-a456-426614174001, quantity=20]
Caused by: com.baeldung.spring.cloud.aws.sqs.acknowledgement.exception.OutOfStockException: Product with id 123e4567-e89b-12d3-a456-426614174001 is out of stock. Quantity requested: 20
INFO 2835 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Stopping container no-retries-order-processing-container
INFO 2835 --- [main] a.c.s.l.AbstractMessageListenerContainer : Container no-retries-order-processing-container stopped
INFO 2835 --- [main] a.s.a.OrderProcessingApplicationLiveTest : Checking for messages in queue order_processing_no_retries_queue
INFO 2835 --- [main] a.s.a.OrderProcessingApplicationLiveTest : No messages found in queue order_processing_no_retries_queue
8. Conclusion
In this article, we used an event-driven scenario to showcase the three acknowledgment modes provided by Spring Cloud AWS v3 SQS integration: ON_SUCCESS (the default), MANUAL, and ALWAYS.
We leveraged the auto-configured settings and used the @SqsListener annotation properties to switch between modes. We also created live tests to assert the behavior using Testcontainers and LocalStack.
As usual, the complete code used in this article is available over on GitHub.
