SQS: 3.4.0+ Observability breaks Visibility Timeout Extension

[marcotesche]

Type: Bug

Component: SQS

Describe the bug
This bug affects Spring Cloud AWS v3.4.0 and newer.
The visibility timeout extension for individual messages in a FIFO message group batch no longer works correctly.

The OriginalBatchMessageVisibilityExtendingInterceptor tracks in-flight messages in a collection and removes them as they are processed using collection.remove(message).
However, since v3.4.0, the immutable Message object is modified after being added to the tracking collection to include a micrometer.observation header.
This new Message object is no longer equal to the one originally stored in the interceptor's collection. Consequently, the remove() call fails silently.

As a result, the interceptor's internal count of pending messages never decreases and the visibility timeout for the remaining messages in the batch is never extended.

This leads to message redeliveries if the total batch processing time exceeds the initial visibility timeout, even if the individual messages are processed within the visibility timeout.

Sample
The provided JUnit test demonstrates the issue. It configures a listener to process 10 messages from a single FIFO group. Each message takes 2 seconds to process, and the visibility timeout is set to 5 seconds.

Expected Behaviour: The test should pass. The visibility timeout should be extended after each message is processed, allowing all 10 messages to be processed successfully without being redelivered. The final processed count should be 10.

Actual Behavior: With version 3.4.0, the test fails. After 5 seconds, unprocessed but already received messages become visible again and are redelivered multiple times, causing the final count of processed messages to be 30 (10+8+6+4+2).

With version 3.3.1 - before introducing observability - the test passes.

Note on LocalStack:
The sample test uses LocalStack for convenience. Be aware that LocalStack has its own inaccuracies in emulating SQS FIFO visibility timeouts (see e.g. related LocalStack issue: localstack/localstack#12457 ).
This causes the test to fail even on a Spring Cloud AWS version without this issue. With a real SQS queue, the test passes on 3.3.1.

However, you can still verify the issue by observing the logs.

• Set the SQS log level to TRACE logging.level.io.awspring.cloud.sqs=TRACE
• Run the test against Spring Cloud AWS 3.3.1. You will see the message "Changing visibility of messages" after each message is processed.
• Run the test against version 3.4.0. This log message will be absent.

For both versions, you should see the message "Adding visibility interceptor for messages" after the initial poll.

build.gradle.kts

plugins {
    java
    id("org.springframework.boot") version "3.5.3"
    id("io.spring.dependency-management") version "1.1.7"
}

group = "com.example"
version = "0.0.1-SNAPSHOT"

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(21)
    }
}

repositories {
    mavenCentral()
}

val springCloudAwsVersion = "3.4.0"
//val springCloudAwsVersion = "3.3.1"

dependencies {
    implementation("org.springframework.boot:spring-boot-starter")
    implementation("io.awspring.cloud:spring-cloud-aws-starter-sqs:$springCloudAwsVersion")
    implementation("io.awspring.cloud:spring-cloud-aws-testcontainers:$springCloudAwsVersion")

    testImplementation("org.springframework.boot:spring-boot-starter-test")
    testImplementation("org.springframework.boot:spring-boot-testcontainers")
    testImplementation("org.testcontainers:junit-jupiter")
    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
}

tasks.withType<Test> {
    useJUnitPlatform()
}

SpringCloudAwsSqsBugVisibilityTimeoutApplicationTests.java

package com.example.springcloudawssqsbugvisibilitytimeout;

import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory;
import io.awspring.cloud.testcontainers.LocalstackAwsClientFactory;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.assertEquals;

@SpringBootTest
class SpringCloudAwsSqsBugVisibilityTimeoutApplicationTests {
    private static final Logger log = LoggerFactory.getLogger(SpringCloudAwsSqsBugVisibilityTimeoutApplicationTests.class);

    private static final String QUEUE_NAME = "test-queue-visibility-timeout.fifo";

    private static final int NUMBER_OF_MESSAGES = 10;
    private static final int VISIBILITY_TIMEOUT_SECONDS = 5;
    private static final int MESSAGE_PROCESSING_SLEEP_DURATION_MS = 2_000;

    /**
     * Note: Test currently fails with localstack even when spring cloud aws is extending the visibility timeout.
     * Use a real AWS SQS queue to get a passing test with spring cloud aws 3.3.1
     */
    @Container
    @ServiceConnection
    private static final LocalStackContainer localStackContainer = new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.6.0"))
            .withServices(LocalStackContainer.Service.SQS);

    @Test
    void testSpringCloudAwsSqsHandlesVisibilityTimeoutResetAfterEachMessageOfGroup() {
        final LocalstackAwsClientFactory factory = new LocalstackAwsClientFactory(localStackContainer);
        final SqsAsyncClient sqsAsyncClient = factory.create(SqsAsyncClient.builder());

        final String queueUrl = setupQueue(sqsAsyncClient);
        final AtomicInteger messagesProcessedCount = new AtomicInteger(0);
        final AtomicInteger messagesInProgressCount = new AtomicInteger(0);

        sendMessagesWithSameGroupId(sqsAsyncClient, queueUrl);

        SqsMessageListenerContainerFactory
                .builder()
                .sqsAsyncClient(sqsAsyncClient)
                .messageListener(message -> {
                    var currentInProgress = messagesInProgressCount.incrementAndGet();
                    log.info("Received message (Currently in Progress: {}) {}", currentInProgress, message);

                    sleep(MESSAGE_PROCESSING_SLEEP_DURATION_MS);

                    final int currentCount = messagesProcessedCount.incrementAndGet();
                    messagesInProgressCount.decrementAndGet();
                    log.info("Processed message #{} {}", currentCount, message);
                })
                .configure(config -> config.messageVisibility(Duration.ofSeconds(VISIBILITY_TIMEOUT_SECONDS)))
                .build()
                .createContainer(queueUrl)
                .start();

        waitForCountToNotChangeAnymore(messagesProcessedCount);

        log.info("Total messages received: {}", messagesProcessedCount.get());

        assertEquals(NUMBER_OF_MESSAGES, messagesProcessedCount.get());
    }

    private void waitForCountToNotChangeAnymore(AtomicInteger count) {
        int previousCount = count.get();

        while (true) {
            sleep(MESSAGE_PROCESSING_SLEEP_DURATION_MS * 2);
            int currentCount = count.get();
            if (currentCount == previousCount) {
                break;
            }
            previousCount = currentCount;
        }
    }

    private String setupQueue(SqsAsyncClient sqsAsyncClient) {
        final var response = sqsAsyncClient.createQueue(CreateQueueRequest.builder()
                        .queueName(QUEUE_NAME)
                        .attributes(Map.of(
                                QueueAttributeName.FIFO_QUEUE, "true",
                                QueueAttributeName.VISIBILITY_TIMEOUT, String.valueOf(VISIBILITY_TIMEOUT_SECONDS)
                        ))
                        .build())
                .join();

        log.info("Created queue {}", response);

        return response.queueUrl();
    }

    private void sendMessagesWithSameGroupId(SqsAsyncClient sqsAsyncClient, String queueUrl) {
        final List<SendMessageBatchRequestEntry> messages = new ArrayList<>();

        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
            messages.add(SendMessageBatchRequestEntry.builder()
                    .messageDeduplicationId(UUID.randomUUID().toString())
                    .messageBody("{}")
                    .messageGroupId("test-message-group")
                    .id(UUID.randomUUID().toString())
                    .build());
        }

        sqsAsyncClient.sendMessageBatch(
                SendMessageBatchRequest.builder()
                        .entries(messages)
                        .queueUrl(queueUrl)
                        .build()
        ).join();
    }

    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

[tomazfernandes]

Hey @marcotesche, thanks for bringing this up, and for the thorough analysis.

Relying on the message instance for identity does seem brittle. A Map keyed by message id might be a more robust way to track the batch, though definitely open to other ideas.

What do you think?

Would you be up for opening a PR with the fix? Ideally, we’d also include an integration test that reproduces the bug.

Let me know your thoughts, thanks.

[marcotesche]

Hey @tomazfernandes,

thanks for getting back to me so fast. Absolutely, happy to help. I've submitted a PR using your suggestion to track the batch - it's a great straightforward solution.

I also considered a more involved refactor, that would also allow users to - for example - implement interceptors that manage periodic visibility timeout heartbeats for FIFO batches. But that's a bigger conversation for another time.