Spring Cloud AWS

Spring Cloud AWS provides full support for FIFO SQS queues (queues ending in .fifo).

Key FIFO Behaviors

• Ordering: Messages are polled with a receiveRequestAttemptId. The batch is split by MessageGroupId. Messages within a group are processed in order, while different groups are processed in parallel.
• Batching Strategy: To receive messages from multiple groups in a single batch, set fifoBatchGroupingStrategy to PROCESS_MULTIPLE_GROUPS_IN_SAME_BATCH in SqsContainerOptions.
• Failure Handling: If processing fails for a message, subsequent messages in the same group are discarded and will be served again after the message visibility expires.
• Visibility: If visibility is set via @SqsListener or SqsContainerOptions, it is extended for all messages in the message group before processing starts.

Important: A MessageListenerContainer can support either Standard queues or FIFO queues, but not both simultaneously.

AWS Kinesis does not support consumer groups natively. The Kinesis Binder implements this using a MetadataStore (for shard checkpoints) and a LockRegistry (to ensure exclusive shard access).

To achieve a highly available (HA) consumer group:

1. Ensure all instances use a shared DynamoDbMetadataStore and DynamoDbLockRegistry.
2. Use the same group name for the channel across all instances via spring.cloud.stream.bindings.<bindingTarget>.group.

Note on Distribution: Even distribution across instances is not guaranteed. A single instance might pick up all shards. To improve throughput, configure consumer concurrency using spring.cloud.stream.bindings.<bindingTarget>.consumer.concurrency.

For static shard distribution, you can manually balance shards by setting:

• spring.cloud.stream.instanceCount to the total number of instances.
• spring.cloud.stream.instanceIndex to the current instance's index.

Spring Cloud AWS provides S3Resource objects, allowing you to treat S3 objects as standard Spring Resource abstractions using the s3:// protocol. You can access them via @Value or the ApplicationContext.

Key behaviors:

• URL Format: s3://[S3_BUCKET_NAME]/[FILE_NAME]
• Trailing Slashes: As of version 4.1.0, trailing slashes are preserved. s3://my-bucket/path/ resolves to the prefix path/. This allows using S3 prefixes as Spring static resource locations.
• Writing: To write to a resource, it must be cast to WritableResource or S3Resource (to set metadata).

// Using @Value
@Value("s3://my-bucket/file.txt")
private Resource s3Resource;

// Using ApplicationContext
Resource res = SpringApplication.run(...).getResource("s3://my-bucket/file.txt");

// Writing with metadata
ObjectMetadata metadata = ObjectMetadata.builder()
    .contentType("application/json")
    .serverSideEncryption(ServerSideEncryption.AES256)
    .build();
((S3Resource) s3Resource).setObjectMetadata(metadata);

try (OutputStream os = s3Resource.getOutputStream()) {
    os.write("content".getBytes());
}

Spring Cloud AWS uses software.amazon.awssdk.auth.credentials.AwsCredentialsProvider to authenticate calls to AWS. You can configure credentials in three ways:

1. DefaultCredentialsProvider: The starter auto-configures this, which searches in the following order:

   • Java system properties (aws.accessKeyId, aws.secretAccessKey)
   • Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
   • Web identity token credentials
   • Credential profiles file (~/.aws/credentials)
   • Amazon EC2 container credentials (AWS_CONTAINER_CREDENTIALS_RELATIVE_URI)
   • EC2 instance profile credentials

2. StsWebIdentityTokenFileCredentialsProvider: Recommended for EKS/Kubernetes environments to assume IAM roles via web identity tokens. Requires the software.amazon.awssdk:sts dependency.

3. Custom AwsCredentialsProvider: Define your own bean to override auto-configuration.

@Configuration
class CustomCredentialsProviderConfiguration {

    @Bean
    public AwsCredentialsProvider customAwsCredentialsProvider() {
        return new CustomAWSCredentialsProvider();
    }
}

The runtime execution of an SQS listener follows a composable four-stage pipeline. Each stage can be extended or swapped by implementing specific interfaces:

• Ingress (MessageSource): Polls SQS for messages and converts them into Spring Message objects. It uses a BackPressureHandler to manage in-flight capacity.
• Dispatch (MessageSink): Routes messages to the processing pipeline. Common implementations include:
  • FanOutMessageSink: For single-message dispatch.
  • BatchMessageSink: For batch processing.
  • OrderedMessageSink: For maintaining order.
  • MessageGroupingSinkAdapter: For FIFO queue grouping.
• Processing (MessageProcessingPipeline): The core logic chain consisting of:
  • MessageInterceptor: Hooks for logic before or after processing.
  • MessageListener: The component that actually invokes your @SqsListener method.
  • ErrorHandler: Manages failures during processing.
  • AcknowledgementHandler: Triggers the deletion of the message from SQS.
• Acknowledgement (AcknowledgementProcessor): Performs the actual deletion of processed messages from SQS and notifies an AcknowledgementResultCallback upon success or failure.

In SQS, acknowledging a message is equivalent to deleting it. You can configure acknowledgement behavior via SqsContainerOptions or the @SqsListener annotation using AcknowledgementMode:

• ON_SUCCESS: Acknowledges after successful processing.
• ALWAYS: Acknowledges after processing returns success or error.
• MANUAL: Framework does not acknowledge automatically; you must use Acknowledgement or BatchAcknowledgement objects in your listener.

Batching Options: Use acknowledgementInterval and acknowledgementThreshold to batch acknowledgements.

• Immediate Acknowledging: Set both to Duration.ZERO and 0 respectively. Messages are acknowledged sequentially after processing.

Ordering Options:

• PARALLEL: Many calls can be made in parallel.
• ORDERED: One batch is executed after the previous one completes (ensures FIFO for batching).
• ORDERED_BY_GROUP: Ensures FIFO ordering per message group (only for FIFO queues).

@Bean
SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient) {
    return SqsMessageListenerContainerFactory
            .builder()
            .configure(options -> options
                    .acknowledgementMode(AcknowledgementMode.ALWAYS)
                    .acknowledgementInterval(Duration.ofSeconds(3))
                    .acknowledgementThreshold(5)
                    .acknowledgementOrdering(AcknowledgementOrdering.ORDERED)
            )
            .sqsAsyncClient(sqsAsyncClient)
            .build();
}

Once imported, secrets are available in the Spring environment.

JSON Secrets

When a SecretString contains JSON, all top-level keys are added as individual properties. For a secret containing {"username": "saanvi", "password": "EXAMPLE-PASSWORD"}, you can access them via @Value or @ConfigurationProperties.

Plain Text Secrets

For plain text secrets, the value is retrieved by referencing the secret name (or the key assigned to it). For example, if a secret named /secrets/prod/jdbc-url contains a JDBC URL, you can use it in your properties file.

Adding a Property Prefix

To prevent collisions, you can add a prefix to all keys within a secret using the ?prefix= syntax in the import string. Note: If you want a dot between the prefix and the key, include a trailing dot in the prefix value.

Example: spring.config.import=aws-secretsmanager:/secrets/database-secrets?prefix=db. will turn username into db.username.

// Accessing JSON keys
@Value("${username}")
private String username;

@Value("${password}")
private String password;