Introduction to Spring Cloud AWS 3.0 – SQS Integration

1. Overview

Spring Cloud AWS is a project that aims to simplify interacting with AWS services. Simple Queue Service (SQS) is an AWS solution for sending and receiving asynchronous messages in a scalable way.

In this tutorial, we’ll reintroduce the Spring Cloud AWS SQS integration, which has been completely rewritten for Spring Cloud AWS 3.0.

The framework provides familiar Spring abstractions for handling SQS queues, such as SqsTemplate and the @SqsListener annotation.

We’ll walk through an event-driven scenario with examples for sending and receiving messages, and show strategies for setting up integration tests with Testcontainers, a tool for managing disposable docker containers, and LocalStack, which simulates an AWS-like environment locally for testing our logic.

2. Dependencies

The Spring Cloud AWS Bill of Materials (BOM) ensures compatible versions between projects. It declares versions for many dependencies, including Spring Boot, and should be used instead of Spring Boot’s own BOM.

Here’s how to import it in our pom.xml file:

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.awspring.cloud</groupId>
            <artifactId>spring-cloud-aws</artifactId>
            <version>3.2.0</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

The main dependency we’ll need is SQS Starter, which contains all SQS-related classes for the project. The SQS integration has no dependency on Spring Boot and can be used standalone in any standard Java application:

<dependency>
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-aws-starter-sqs</artifactId>
</dependency>

For Spring Boot applications such as the one we’re building in this tutorial, we should add the project’s Core Starter, as it allows us to leverage Spring Boot’s auto-configuration for SQS, and AWS configuration such as credentials and region:

<dependency>
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-aws-starter</artifactId>
</dependency>

3. Setting up a Local Test Environment

In this section, we’ll walk through setting up a LocalStack environment with Testcontainers to test our code in our local environment. Note that the examples in this tutorial can also be executed targeting AWS directly.

3.1. Dependencies

For running LocalStack and TestContainers with JUnit 5, we’ll need two additional dependencies:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>localstack</artifactId>
    <scope>test</scope>
</dependency>

Let’s also include the awaitility library to help us assert the asynchronous message consumption:

<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

3.2. Configuration

Next, we’ll configure our local test environment using Testcontainers and LocalStack. We’ll create a SqsLiveTestConfiguration class:

@Configuration
public class SqsLiveTestConfiguration {
    private static final String LOCAL_STACK_VERSION = "localstack/localstack:3.4.0";

    @Bean
    @ServiceConnection
    LocalStackContainer localStackContainer() {
        return new LocalStackContainer(DockerImageName.parse(LOCAL_STACK_VERSION));
    }
}

4. Setting Up the Queue Names

We can set up the queue names by leveraging Spring Boot’s application.yml property mechanism.

For this tutorial, we’ll create three queues:

events:
  queues:
    user-created-by-name-queue: user_created_by_name_queue
    user-created-record-queue: user_created_record_queue
    user-created-event-type-queue: user_created_event_type_queue

Let’s create a POJO to represent these properties:

@ConfigurationProperties(prefix = "events.queues")
public class EventQueuesProperties {

    private String userCreatedByNameQueue;
    private String userCreatedRecordQueue;
    private String userCreatedEventTypeQueue;

    // getters and setters
}

Finally, we need to use the @EnableConfigurationProperties annotation in a @Configuration annotated class, or the main Spring Application class, to let Spring Boot know we want to populate it with our application.yml properties:

@SpringBootApplication
@EnableConfigurationProperties(EventQueuesProperties.class)
public class SpringCloudAwsApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringCloudAwsApplication.class, args);
    }
}

Now, we’re ready to inject either the values themselves or the POJO when we need the queue names.

By default, Spring Cloud AWS SQS will create the queues if they’re not found, which helps us quickly set up dev environments. In production, the application should not have permission to create queues, so it will fail to start if a queue is not found. The framework can also be configured to explicitly fail if a queue is not found instead.

5. Sending and Receiving Messages

There are multiple ways of sending and receiving messages to and from SQS using Spring Cloud AWS. Here, we’ll cover the most common ones, using the SqsTemplate for sending messages and the @SqsListener annotation for receiving them.

5.1. Scenario

In our scenario, we’ll simulate an event-driven application that responds to UserCreatedEvent by saving relevant information in its local repository.

Let’s create a User entity:

public record User(String id, String name, String email) {
}

And let’s create a simple in-memory UserRepository:

@Repository
public class UserRepository {

    private final Map<String, User> persistedUsers = new ConcurrentHashMap<>();

    public void save(User userToSave) {
        persistedUsers.put(userToSave.id(), userToSave);
    }

    public Optional<User> findById(String userId) {
        return Optional.ofNullable(persistedUsers.get(userId));
    }

    public Optional<User> findByName(String name) {
        return persistedUsers.values().stream()
          .filter(user -> user.name().equals(name))
          .findFirst();
    }
}

And finally, let’s create a UserCreatedEvent Java Record class:

public record UserCreatedEvent(String id, String username, String email) {
}

5.2. Setup

To test our scenarios, we’ll create a SpringCloudAwsSQSLiveTest class that extends the BaseSqsIntegrationTest file we created earlier. We’ll autowire three dependencies: the SqsTemplate that’s auto-configured by the framework, the UserRepository so we can assert our message processing worked, and our EventQueuesProperties POJO with the queue names:

public class SpringCloudAwsSQSLiveTest extends BaseSqsIntegrationTest {

    private static final Logger logger = LoggerFactory.getLogger(SpringCloudAwsSQSLiveTest.class);

    @Autowired
    private SqsTemplate sqsTemplate;

    @Autowired
    private UserRepository userRepository;

    @Autowired
    private EventQueuesProperties eventQueuesProperties;

   // ...
}

To contain our listeners, let’s create a UserEventListeners class and declare it as a Spring @Component:

@Component
public class UserEventListeners {

    private static final Logger logger = LoggerFactory.getLogger(UserEventListeners.class);

    public static final String EVENT_TYPE_CUSTOM_HEADER = "eventType";

    private final UserRepository userRepository;

    public UserEventListeners(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    // Our listeners will be added here 
}

5.3. String Payloads

In this first example, we’ll send a message with a String payload, receive it in our listener, and persist it to our repository. We’ll then poll the repository to make sure our application persists the data correctly.

First, let’s create a test for sending the message in our test class:

@Test
void givenAStringPayload_whenSend_shouldReceive() {
    // given
    var userName = "Albert";

    // when
    sqsTemplate.send(to -> to.queue(eventQueuesProperties.getUserCreatedByNameQueue())
      .payload(userName));
    logger.info("Message sent with payload {}", userName);

    // then
    await().atMost(Duration.ofSeconds(3))
      .until(() -> userRepository.findByName(userName)
        .isPresent());
}

We should see a log similar to:

INFO [ main] c.b.s.c.a.sqs.SpringCloudAwsSQSLiveTest : Message sent with payload Albert

And then, note that the test fails because we don’t have the listener for this queue yet.

Let’s set up our listener to consume the message from this queue in our listener class and make the test pass:

@SqsListener("${events.queues.user-created-by-name-queue}")
public void receiveStringMessage(String username) {
    logger.info("Received message: {}", username);
    userRepository.save(new User(UUID.randomUUID()
      .toString(), username, null));
}

Now, when we run the test, we should see the result in the log:

INFO [ntContainer#0-1] c.b.s.cloud.aws.sqs.UserEventListeners : Received message: Albert

And the test passes.

Note that we’re using Spring’s properties-resolving capabilities to fetch the queue name from the application.yml we created earlier.

5.4. POJO and Record Payloads

Now that we’ve sent and received a String payload, let’s set up a scenario with a Java Record, the UserCreatedEvent we created earlier.

First, let’s write our failing test:

@Test
void givenARecordPayload_whenSend_shouldReceive() {
    // given
    String userId = UUID.randomUUID()
      .toString();
    var payload = new UserCreatedEvent(userId, "John", "[email protected]");

    // when
    sqsTemplate.send(to -> to.queue(eventQueuesProperties.getUserCreatedRecordQueue())
      .payload(payload));

    // then
    logger.info("Message sent with payload: {}", payload);
    await().atMost(Duration.ofSeconds(3))
      .until(() -> userRepository.findById(userId)
        .isPresent());
}

We should see a log similar to this before the test fails:

INFO [ main] c.b.s.c.a.sqs.SpringCloudAwsSQSLiveTest : Message sent with payload: UserCreatedEvent[id=67f52cf6-c750-4200-9a02-345bda0516f8, username=John, [email protected]]

Now, let’s create the corresponding listener to make the test pass:

@SqsListener("${events.queues.user-created-record-queue}")
public void receiveRecordMessage(UserCreatedEvent event) {
    logger.info("Received message: {}", event);
    userRepository.save(new User(event.id(), event.username(), event.email()));
}

We’ll see the output noting the message was received, and the test passes:

INFO [ntContainer#1-1] c.b.s.cloud.aws.sqs.UserEventListeners   : Received message: UserCreatedEvent[id=2d66df3d-2dbd-4aed-8fc0-ddd08416ed12, username=John, [email protected]]

The framework will auto-configure any ObjectMapper bean available in the Spring Context to handle the serialization and deserialization of messages. We can configure our own ObjectMapper and customize serialization in several ways, but that’s beyond the scope of this tutorial.

5.5. Spring Message and Headers

In this last scenario, we’ll send a Record with a custom header and receive the message as a Spring Message instance, as well as both the custom header we added and a standard SQS header in the method signature. The framework automatically converts all SQS message attributes to message headers, including any provided by the user.

Let’s create the failing test first:

@Test
void givenCustomHeaders_whenSend_shouldReceive() {
    // given
    String userId = UUID.randomUUID()
      .toString();
    var payload = new UserCreatedEvent(userId, "John", "[email protected]");
    var headers = Map.<String, Object> of(EVENT_TYPE_CUSTOM_HEADER, "UserCreatedEvent");

    // when
    sqsTemplate.send(to -> to.queue(eventQueuesProperties.getUserCreatedEventTypeQueue())
      .payload(payload)
      .headers(headers));

    // then
    logger.info("Sent message with payload {} and custom headers: {}", payload, headers);
    await().atMost(Duration.ofSeconds(3))
      .until(() -> userRepository.findById(userId)
        .isPresent());
}

The test should generate a log similar to this before failing:

INFO [ main] c.b.s.c.a.sqs.SpringCloudAwsSQSLiveTest  : Sent message with payload UserCreatedEvent[id=575de854-82de-44e4-8dfe-8fdc9f6ae4a1, username=John, [email protected]] and custom headers: {eventType=UserCreatedEvent}

Now, let’s add the corresponding listener to make the test pass:

@SqsListener("${events.queues.user-created-event-type-queue}")
public void customHeaderMessage(Message<UserCreatedEvent> message, @Header(EVENT_TYPE_CUSTOM_HEADER) String eventType,
    @Header(SQS_APPROXIMATE_FIRST_RECEIVE_TIMESTAMP) Long firstReceive) {
    logger.info("Received message {} with event type {}. First received at approximately {}.", message, eventType, firstReceive);
    UserCreatedEvent payload = message.getPayload();
    userRepository.save(new User(payload.id(), payload.username(), payload.email()));
}

And we’ll see the output when we re-run our test, indicating success:

INFO [ntContainer#2-1] c.b.s.cloud.aws.sqs.UserEventListeners   : Received message GenericMessage [payload=UserCreatedEvent[id=575de854-82de-44e4-8dfe-8fdc9f6ae4a1, username=John, [email protected]], headers=...

In this example, we’re receiving a Message with the deserialized UserCreatedEvent record as the payload and two headers. To ensure consistency throughout the project, we should use the SqsHeader class constants for retrieving SQS standard headers.

6. Conclusion

In this article, we used an event-driven scenario to go through different examples for sending and receiving messages with Spring Cloud AWS SQS 3.0.

We set up a local environment with LocalStack and TestContainers and configured the framework to use the proper local configuration for our integration tests.