Aws Sdk Java V2 Messaging

AWS SDK for Java 2.x - Messaging (SQS & SNS)

Overview

Provides patterns for SQS queues and SNS topics with AWS SDK for Java 2.x: client setup, queue management, message operations, subscriptions, and Spring Boot integration.

When to Use

• Setting up SQS queues (standard or FIFO) for message buffering
• Implementing pub/sub with SNS topics and subscriptions
• Processing messages from SQS queues with long polling
• Configuring dead letter queues (DLQ) for error handling
• Integrating AWS messaging with Spring Boot applications
• Building event-driven architectures with SQS/SNS

Examples

Quick Setup

Dependencies:

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sns</artifactId>
</dependency>

Client Configuration:

SqsClient sqsClient = SqsClient.builder()
    .region(Region.US_EAST_1)
    .credentialsProvider(DefaultCredentialsProvider.create())
    .build();

SnsClient snsClient = SnsClient.builder()
    .region(Region.US_EAST_1)
    .build();

SQS Operations

Create and Send Message:

String queueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue")
    .build()).queueUrl();

String messageId = sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(queueUrl)
    .messageBody("Hello, SQS!")
    .build()).messageId();

Receive and Delete Message:

ReceiveMessageResponse response = sqsClient.receiveMessage(ReceiveMessageRequest.builder()
    .queueUrl(queueUrl)
    .maxNumberOfMessages(10)
    .waitTimeSeconds(20)
    .build());

response.messages().forEach(message -> {
    processMessage(message.body());
    sqsClient.deleteMessage(DeleteMessageRequest.builder()
        .queueUrl(queueUrl)
        .receiptHandle(message.receiptHandle())
        .build());
});

FIFO Queue:

Map<QueueAttributeName, String> attributes = Map.of(
    QueueAttributeName.FIFO_QUEUE, "true",
    QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true"
);

String fifoQueueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue.fifo")
    .attributes(attributes)
    .build()).queueUrl();

sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(fifoQueueUrl)
    .messageBody("Order #12345")
    .messageGroupId("orders")
    .messageDeduplicationId(UUID.randomUUID().toString())
    .build());

SNS Operations

Create Topic and Publish:

String topicArn = snsClient.createTopic(CreateTopicRequest.builder()
    .name("my-topic")
    .build()).topicArn();

snsClient.publish(PublishRequest.builder()
    .topicArn(topicArn)
    .subject("Test Notification")
    .message("Hello, SNS!")
    .build());

SNS to SQS Subscription:

String queueArn = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder()
    .queueUrl(queueUrl)
    .attributeNames(QueueAttributeName.QUEUE_ARN)
    .build()).attributes().get(QueueAttributeName.QUEUE_ARN);

snsClient.subscribe(SubscribeRequest.builder()
    .protocol("sqs")
    .endpoint(queueArn)
    .topicArn(topicArn)
    .build());

Spring Boot Integration

@Service
@RequiredArgsConstructor
public class OrderNotificationService {
    private final SnsClient snsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sns.order-topic-arn}")
    private String orderTopicArn;

    public void sendOrderNotification(Order order) throws JsonProcessingException {
        snsClient.publish(PublishRequest.builder()
            .topicArn(orderTopicArn)
            .subject("New Order Received")
            .message(objectMapper.writeValueAsString(order))
            .messageAttributes(Map.of(
                "orderType", MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue(order.getType())
                    .build()))
            .build());
    }
}

Instructions

Implement Message Processing (with Validation)

1. Create queues/topics with appropriate configuration 2. Send messages and validate messageId is returned 3. Receive messages with long polling (waitTimeSeconds: 20) 4. Process messages - validate payload before processing 5. Delete messages only after successful processing - verify deletion response 6. Check DLQ periodically for failed messages using redrivePolicy 7. Verify delivery - monitor CloudWatch NumberOfMessagesSent metric

Validation Checklist:

// After send
if (messageId == null || messageId.isEmpty()) {
    throw new MessagingException("Message send failed - no messageId returned");
}

// After receive
if (response.messages().isEmpty()) {
    log.debug("No messages available - normal with long polling");
}

// After delete
if (!deleteResponse.sdkHttpResponse().isSuccessful()) {
    throw new MessagingException("Message deletion failed");
}

Setup Credentials

export AWS_ACCESS_KEY_ID=your-access-key
export AWS_SECRET_ACCESS_KEY=your-secret-key
export AWS_REGION=us-east-1

Monitor and Debug

• CloudWatch metrics: ApproximateNumberOfMessages, NumberOfMessagesSent, NumberOfMessagesReceived
• Enable SDK logging: software.amazon.awssdk at DEBUG level
• Use X-Ray for distributed tracing

Best Practices

SQS:

• Use long polling (20-40s) to reduce empty responses and costs
• Always delete messages after successful processing
• Implement idempotent processing for duplicate handling
• Configure DLQ (redrivePolicy) for failed messages
• Use FIFO queues when order matters (300 msg/sec limit)

SNS:

• Use filter policies to reduce unnecessary deliveries
• Keep messages under 256KB
• Implement retry with exponential backoff
• Monitor NumberOfNotificationFailed metric

General:

• Use IAM roles over static credentials
• Reuse clients (they are thread-safe)
• Test with LocalStack or Testcontainers

Detailed References

• references/detailed-sqs-operations.md
• references/detailed-sns-operations.md
• references/spring-boot-integration.md
• references/aws-official-documentation.md

Constraints and Warnings

• Message Size: Maximum 256KB for SQS and SNS
• Visibility Timeout: Undeleted messages reappear after timeout - always delete after processing
• Input Validation: Sanitize message body before processing - messages may contain untrusted payloads
• FIFO Naming: Must end with .fifo suffix
• FIFO Throughput: 300 msg/sec per queue (use partitioning for higher throughput)
• Message Retention: SQS retains messages max 14 days
• DLQ Required: Configure dead letter queue to prevent message loss
• Region-Specific: SQS queues are region-specific; cross-region requires SNS

AWS SQS & SNS Official Documentation Reference

This file contains reference information extracted from official AWS resources for the AWS SDK for Java 2.x messaging patterns.

Source Documents

• AWS Java SDK v2 Examples - SQS
• AWS Java SDK v2 Examples - SNS
• AWS SQS Developer Guide
• AWS SNS Developer Guide

Amazon SQS Reference

Core Operations

• CreateQueue - Create new SQS queue
• DeleteMessage - Delete individual message from queue
• ListQueues - List available queues
• ReceiveMessage - Receive messages from queue
• SendMessage - Send message to queue
• SendMessageBatch - Send multiple messages to queue

Advanced Features

• Large Message Handling - Use S3 for messages larger than 256KB
• Batch Operations - Process multiple messages efficiently
• Long Polling - Reduce empty responses with waitTimeSeconds
• Visibility Timeout - Control message visibility during processing
• Dead Letter Queues (DLQ) - Handle failed messages
• FIFO Queues - Ensure message ordering and deduplication

Java SDK v2 Key Classes

// Core clients and models
software.amazon.awssdk.services.sqs.SqsClient
software.amazon.awssdk.services.sqs.model.*
software.amazon.awssdk.services.sqs.model.QueueAttributeName

Amazon SNS Reference

Core Operations

• CreateTopic - Create new SNS topic
• Publish - Send message to topic
• Subscribe - Subscribe endpoint to topic
• ListSubscriptions - List topic subscriptions
• Unsubscribe - Remove subscription

Advanced Features

• Platform Endpoints - Mobile push notifications
• SMS Publishing - Send SMS messages
• FIFO Topics - Ordered message delivery with deduplication
• Filter Policies - Filter messages based on attributes
• Message Attributes - Enrich messages with metadata
• DLQ for Subscriptions - Handle failed deliveries

Java SDK v2 Key Classes

// Core clients and models
software.amazon.awssdk.services.sns.SnsClient
software.amazon.awssdk.services.sns.model.*
software.amazon.awssdk.services.sns.model.MessageAttributeValue

Best Practices from AWS

SQS Best Practices

1. Use Long Polling: Set waitTimeSeconds (10-40 seconds) to reduce empty responses 2. Batch Operations: Use SendMessageBatch for efficiency 3. Visibility Timeout: Set appropriately based on processing time 4. Handle Duplicates: Implement idempotent processing for retries 5. Monitor Queue Depth: Use CloudWatch for monitoring 6. Implement DLQ: Route failed messages for analysis

SNS Best Practices

1. Use Filter Policies: Reduce noise by filtering messages 2. Message Attributes: Add metadata for routing decisions 3. Retry Logic: Handle transient failures gracefully 4. Monitor Failed Deliveries: Set up CloudWatch alarms 5. Security: Use IAM policies for access control 6. FIFO Topics: Use when order and deduplication are critical

Error Handling Patterns

Common SQS Errors

• QueueDoesNotExistException: Verify queue URL
• MessageNotInflightException: Check message visibility
• OverLimitException: Implement backoff/retry logic
• InvalidAttributeValueException: Validate queue attributes

Common SNS Errors

• NotFoundException: Verify topic ARN
• InvalidParameterException: Validate subscription parameters
• InternalFailureException: Implement retry logic
• AuthorizationErrorException: Check IAM permissions

Integration Patterns

Spring Boot Integration

• Use @Service classes for business logic
• Inject SqsClient and SnsClient via constructor injection
• Configure clients with @Configuration beans
• Use @Value for externalizing configuration

Testing Strategies

• Use LocalStack for local development
• Mock AWS services with Mockito for unit tests
• Integrate with Testcontainers for integration tests
• Test idempotent operations thoroughly

Configuration Options

SQS Configuration

SqsClient sqsClient = SqsClient.builder()
    .region(Region.US_EAST_1)
    .build();

SNS Configuration

SnsClient snsClient = SnsClient.builder()
    .region(Region.US_EAST_1)
    .build();

Advanced Configuration

• Override endpoint for local development
• Configure custom credentials provider
• Set custom HTTP client
• Configure retry policies

Monitoring and Observability

SQS Metrics

• ApproximateNumberOfMessagesVisible
• ApproximateNumberOfMessagesNotVisible
• ApproximateNumberOfMessagesDelayed
• SentMessages
• ReceiveCalls

SNS Metrics

• NumberOfNotifications
• PublishSuccess
• PublishFailed
• SubscriptionConfirmation
• SubscriptionConfirmationFailed

Security Considerations

IAM Permissions

• Grant least privilege access
• Use IAM roles for EC2/ECS
• Implement resource-based policies
• Use condition keys for fine-grained control

Data Protection

• Encrypt sensitive data in messages
• Use KMS for message encryption
• Implement message signing
• Secure endpoints with HTTPS

Detailed SNS Operations Reference

Topic Management

Create Standard Topic

public String createTopic(SnsClient snsClient, String topicName) {
    CreateTopicRequest request = CreateTopicRequest.builder()
        .name(topicName)
        .build();

    CreateTopicResponse response = snsClient.createTopic(request);
    return response.topicArn();
}

Create FIFO Topic

public String createFifoTopic(SnsClient snsClient, String topicName) {
    Map<String, String> attributes = new HashMap<>();
    attributes.put("FifoTopic", "true");
    attributes.put("ContentBasedDeduplication", "true");

    CreateTopicRequest request = CreateTopicRequest.builder()
        .name(topicName + ".fifo")
        .attributes(attributes)
        .build();

    CreateTopicResponse response = snsClient.createTopic(request);
    return response.topicArn();
}

Topic Operations

public List<Topic> listTopics(SnsClient snsClient) {
    return snsClient.listTopics().topics();
}

public String getTopicArn(SnsClient snsClient, String topicName) {
    return snsClient.createTopic(CreateTopicRequest.builder()
        .name(topicName)
        .build()).topicArn();
}

Message Publishing

Publish Basic Message

public String publishMessage(SnsClient snsClient, String topicArn, String message) {
    PublishRequest request = PublishRequest.builder()
        .topicArn(topicArn)
        .message(message)
        .build();

    PublishResponse response = snsClient.publish(request);
    return response.messageId();
}

Publish with Subject

public String publishMessageWithSubject(SnsClient snsClient,
                                        String topicArn,
                                        String subject,
                                        String message) {
    PublishRequest request = PublishRequest.builder()
        .topicArn(topicArn)
        .subject(subject)
        .message(message)
        .build();

    PublishResponse response = snsClient.publish(request);
    return response.messageId();
}

Publish with Attributes

public String publishMessageWithAttributes(SnsClient snsClient,
                                           String topicArn,
                                           String message,
                                           Map<String, String> attributes) {
    Map<String, MessageAttributeValue> messageAttributes = attributes.entrySet().stream()
        .collect(Collectors.toMap(
            Map.Entry::getKey,
            e -> MessageAttributeValue.builder()
                .dataType("String")
                .stringValue(e.getValue())
                .build()));

    PublishRequest request = PublishRequest.builder()
        .topicArn(topicArn)
        .message(message)
        .messageAttributes(messageAttributes)
        .build();

    PublishResponse response = snsClient.publish(request);
    return response.messageId();
}

Publish FIFO Message

public String publishFifoMessage(SnsClient snsClient,
                                 String topicArn,
                                 String message,
                                 String messageGroupId) {
    PublishRequest request = PublishRequest.builder()
        .topicArn(topicArn)
        .message(message)
        .messageGroupId(messageGroupId)
        .messageDeduplicationId(UUID.randomUUID().toString())
        .build();

    PublishResponse response = snsClient.publish(request);
    return response.messageId();
}

Subscription Management

Subscribe Email to Topic

public String subscribeEmail(SnsClient snsClient, String topicArn, String email) {
    SubscribeRequest request = SubscribeRequest.builder()
        .protocol("email")
        .endpoint(email)
        .topicArn(topicArn)
        .build();

    SubscribeResponse response = snsClient.subscribe(request);
    return response.subscriptionArn();
}

Subscribe SQS to Topic

public String subscribeSqs(SnsClient snsClient, String topicArn, String queueArn) {
    SubscribeRequest request = SubscribeRequest.builder()
        .protocol("sqs")
        .endpoint(queueArn)
        .topicArn(topicArn)
        .build();

    SubscribeResponse response = snsClient.subscribe(request);
    return response.subscriptionArn();
}

Subscribe Lambda to Topic

public String subscribeLambda(SnsClient snsClient, String topicArn, String lambdaArn) {
    SubscribeRequest request = SubscribeRequest.builder()
        .protocol("lambda")
        .endpoint(lambdaArn)
        .topicArn(topicArn)
        .build();

    SubscribeResponse response = snsClient.subscribe(request);
    return response.subscriptionArn();
}

Subscription Operations

public List<Subscription> listSubscriptions(SnsClient snsClient, String topicArn) {
    return snsClient.listSubscriptionsByTopic(ListSubscriptionsByTopicRequest.builder()
        .topicArn(topicArn)
        .build()).subscriptions();
}

public void unsubscribe(SnsClient snsClient, String subscriptionArn) {
    snsClient.unsubscribe(UnsubscribeRequest.builder()
        .subscriptionArn(subscriptionArn)
        .build());
}

Detailed SQS Operations Reference

Queue Management

Create Standard Queue

public String createQueue(SqsClient sqsClient, String queueName) {
    CreateQueueRequest request = CreateQueueRequest.builder()
        .queueName(queueName)
        .build();

    CreateQueueResponse response = sqsClient.createQueue(request);
    return response.queueUrl();
}

Create FIFO Queue

public String createFifoQueue(SqsClient sqsClient, String queueName) {
    Map<QueueAttributeName, String> attributes = new HashMap<>();
    attributes.put(QueueAttributeName.FIFO_QUEUE, "true");
    attributes.put(QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true");

    CreateQueueRequest request = CreateQueueRequest.builder()
        .queueName(queueName + ".fifo")
        .attributes(attributes)
        .build();

    CreateQueueResponse response = sqsClient.createQueue(request);
    return response.queueUrl();
}

Queue Operations

public String getQueueUrl(SqsClient sqsClient, String queueName) {
    return sqsClient.getQueueUrl(GetQueueUrlRequest.builder()
        .queueName(queueName)
        .build()).queueUrl();
}

public List<String> listQueues(SqsClient sqsClient) {
    return sqsClient.listQueues().queueUrls();
}

public void purgeQueue(SqsClient sqsClient, String queueUrl) {
    sqsClient.purgeQueue(PurgeQueueRequest.builder()
        .queueUrl(queueUrl)
        .build());
}

Message Operations

Send Basic Message

public String sendMessage(SqsClient sqsClient, String queueUrl, String messageBody) {
    SendMessageRequest request = SendMessageRequest.builder()
        .queueUrl(queueUrl)
        .messageBody(messageBody)
        .build();

    SendMessageResponse response = sqsClient.sendMessage(request);
    return response.messageId();
}

Send Message with Attributes

public String sendMessageWithAttributes(SqsClient sqsClient,
                                        String queueUrl,
                                        String messageBody,
                                        Map<String, String> attributes) {
    Map<String, MessageAttributeValue> messageAttributes = attributes.entrySet().stream()
        .collect(Collectors.toMap(
            Map.Entry::getKey,
            e -> MessageAttributeValue.builder()
                .dataType("String")
                .stringValue(e.getValue())
                .build()));

    SendMessageRequest request = SendMessageRequest.builder()
        .queueUrl(queueUrl)
        .messageBody(messageBody)
        .messageAttributes(messageAttributes)
        .build();

    SendMessageResponse response = sqsClient.sendMessage(request);
    return response.messageId();
}

Send FIFO Message

public String sendFifoMessage(SqsClient sqsClient,
                              String queueUrl,
                              String messageBody,
                              String messageGroupId) {
    SendMessageRequest request = SendMessageRequest.builder()
        .queueUrl(queueUrl)
        .messageBody(messageBody)
        .messageGroupId(messageGroupId)
        .messageDeduplicationId(UUID.randomUUID().toString())
        .build();

    SendMessageResponse response = sqsClient.sendMessage(request);
    return response.messageId();
}

Send Batch Messages

public void sendBatchMessages(SqsClient sqsClient,
                              String queueUrl,
                              List<String> messages) {
    List<SendMessageBatchRequestEntry> entries = IntStream.range(0, messages.size())
        .mapToObj(i -> SendMessageBatchRequestEntry.builder()
            .id(String.valueOf(i))
            .messageBody(messages.get(i))
            .build())
        .collect(Collectors.toList());

    SendMessageBatchRequest request = SendMessageBatchRequest.builder()
        .queueUrl(queueUrl)
        .entries(entries)
        .build();

    SendMessageBatchResponse response = sqsClient.sendMessageBatch(request);

    System.out.println("Successful: " + response.successful().size());
    System.out.println("Failed: " + response.failed().size());
}

Message Processing

Receive Messages with Long Polling

public List<Message> receiveMessages(SqsClient sqsClient, String queueUrl) {
    ReceiveMessageRequest request = ReceiveMessageRequest.builder()
        .queueUrl(queueUrl)
        .maxNumberOfMessages(10)
        .waitTimeSeconds(20) // Long polling
        .messageAttributeNames("All")
        .build();

    ReceiveMessageResponse response = sqsClient.receiveMessage(request);
    return response.messages();
}

Delete Message

public void deleteMessage(SqsClient sqsClient, String queueUrl, String receiptHandle) {
    DeleteMessageRequest request = DeleteMessageRequest.builder()
        .queueUrl(queueUrl)
        .receiptHandle(receiptHandle)
        .build();

    sqsClient.deleteMessage(request);
}

Delete Batch Messages

public void deleteBatchMessages(SqsClient sqsClient,
                                String queueUrl,
                                List<Message> messages) {
    List<DeleteMessageBatchRequestEntry> entries = messages.stream()
        .map(msg -> DeleteMessageBatchRequestEntry.builder()
            .id(msg.messageId())
            .receiptHandle(msg.receiptHandle())
            .build())
        .collect(Collectors.toList());

    DeleteMessageBatchRequest request = DeleteMessageBatchRequest.builder()
        .queueUrl(queueUrl)
        .entries(entries)
        .build();

    sqsClient.deleteMessageBatch(request);
}

Change Message Visibility

public void changeMessageVisibility(SqsClient sqsClient,
                                    String queueUrl,
                                    String receiptHandle,
                                    int visibilityTimeout) {
    ChangeMessageVisibilityRequest request = ChangeMessageVisibilityRequest.builder()
        .queueUrl(queueUrl)
        .receiptHandle(receiptHandle)
        .visibilityTimeout(visibilityTimeout)
        .build();

    sqsClient.changeMessageVisibility(request);
}

Spring Boot Integration Reference

Configuration

Basic Bean Configuration

@Configuration
public class MessagingConfiguration {

    @Bean
    public SqsClient sqsClient() {
        return SqsClient.builder()
            .region(Region.US_EAST_1)
            .build();
    }

    @Bean
    public SnsClient snsClient() {
        return SnsClient.builder()
            .region(Region.US_EAST_1)
            .build();
    }
}

Configuration Properties

# application.yml
aws:
  sqs:
    queue-url: https://sqs.us-east-1.amazonaws.com/123456789012/my-queue
  sns:
    topic-arn: arn:aws:sns:us-east-1:123456789012:my-topic

Service Layer Integration

SQS Message Service

@Service
@RequiredArgsConstructor
public class SqsMessageService {

    private final SqsClient sqsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sqs.queue-url}")
    private String queueUrl;

    public <T> void sendMessage(T message) {
        try {
            String jsonMessage = objectMapper.writeValueAsString(message);

            SendMessageRequest request = SendMessageRequest.builder()
                .queueUrl(queueUrl)
                .messageBody(jsonMessage)
                .build();

            sqsClient.sendMessage(request);

        } catch (Exception e) {
            throw new RuntimeException("Failed to send SQS message", e);
        }
    }

    public <T> List<T> receiveMessages(Class<T> messageType) {
        ReceiveMessageRequest request = ReceiveMessageRequest.builder()
            .queueUrl(queueUrl)
            .maxNumberOfMessages(10)
            .waitTimeSeconds(20)
            .build();

        ReceiveMessageResponse response = sqsClient.receiveMessage(request);

        return response.messages().stream()
            .map(msg -> {
                try {
                    return objectMapper.readValue(msg.body(), messageType);
                } catch (Exception e) {
                    throw new RuntimeException("Failed to parse message", e);
                }
            })
            .collect(Collectors.toList());
    }

    public void deleteMessage(String receiptHandle) {
        DeleteMessageRequest request = DeleteMessageRequest.builder()
            .queueUrl(queueUrl)
            .receiptHandle(receiptHandle)
            .build();

        sqsClient.deleteMessage(request);
    }
}

SNS Notification Service

@Service
@RequiredArgsConstructor
public class SnsNotificationService {

    private final SnsClient snsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sns.topic-arn}")
    private String topicArn;

    public void publishNotification(String subject, Object message) {
        try {
            String jsonMessage = objectMapper.writeValueAsString(message);

            PublishRequest request = PublishRequest.builder()
                .topicArn(topicArn)
                .subject(subject)
                .message(jsonMessage)
                .build();

            snsClient.publish(request);

        } catch (Exception e) {
            throw new RuntimeException("Failed to publish SNS notification", e);
        }
    }
}

Message Listener Pattern

Scheduled Polling

@Service
@RequiredArgsConstructor
public class SqsMessageListener {

    private final SqsClient sqsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sqs.queue-url}")
    private String queueUrl;

    @Scheduled(fixedDelay = 5000)
    public void pollMessages() {
        ReceiveMessageRequest request = ReceiveMessageRequest.builder()
            .queueUrl(queueUrl)
            .maxNumberOfMessages(10)
            .waitTimeSeconds(20)
            .build();

        ReceiveMessageResponse response = sqsClient.receiveMessage(request);

        response.messages().forEach(this::processMessage);
    }

    private void processMessage(Message message) {
        try {
            // Process message
            System.out.println("Processing: " + message.body());

            // Delete message after successful processing
            deleteMessage(message.receiptHandle());

        } catch (Exception e) {
            // Handle error - message will become visible again
            System.err.println("Failed to process message: " + e.getMessage());
        }
    }

    private void deleteMessage(String receiptHandle) {
        DeleteMessageRequest request = DeleteMessageRequest.builder()
            .queueUrl(queueUrl)
            .receiptHandle(receiptHandle)
            .build();

        sqsClient.deleteMessage(request);
    }
}

Pub/Sub Pattern Integration

Configuration for Pub/Sub

@Configuration
@RequiredArgsConstructor
public class PubSubConfiguration {

    private final SnsClient snsClient;
    private final SqsClient sqsClient;

    @Bean
    @DependsOn("sqsClient")
    public String setupPubSub() {
        // Create SNS topic
        String topicArn = snsClient.createTopic(CreateTopicRequest.builder()
            .name("order-events")
            .build()).topicArn();

        // Create SQS queue
        String queueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
            .queueName("order-processor")
            .build()).queueUrl();

        // Get queue ARN
        String queueArn = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder()
            .queueUrl(queueUrl)
            .attributeNames(QueueAttributeName.QUEUE_ARN)
            .build()).attributes().get(QueueAttributeName.QUEUE_ARN);

        // Subscribe SQS to SNS
        snsClient.subscribe(SubscribeRequest.builder()
            .protocol("sqs")
            .endpoint(queueArn)
            .topicArn(topicArn)
            .build());

        return topicArn;
    }
}

Error Handling Patterns

Retry Mechanism

@Service
@RequiredArgsConstructor
public class RetryableSqsService {

    private final SqsClient sqsClient;
    private final RetryTemplate retryTemplate;

    public void sendMessageWithRetry(String queueUrl, String messageBody) {
        retryTemplate.execute(context -> {
            try {
                SendMessageRequest request = SendMessageRequest.builder()
                    .queueUrl(queueUrl)
                    .messageBody(messageBody)
                    .build();

                sqsClient.sendMessage(request);
                return null;
            } catch (Exception e) {
                throw new RetryableException("Failed to send message", e);
            }
        });
    }
}

Testing Integration

LocalStack Configuration

@TestConfiguration
public class LocalStackMessagingConfig {

    @Container
    static LocalStackContainer localstack = new LocalStackContainer(
        DockerImageName.parse("localstack/localstack:3.0"))
        .withServices(
            LocalStackContainer.Service.SQS,
            LocalStackContainer.Service.SNS
        );

    @Bean
    public SqsClient sqsClient() {
        return SqsClient.builder()
            .region(Region.US_EAST_1)
            .endpointOverride(
                localstack.getEndpointOverride(LocalStackContainer.Service.SQS))
            .credentialsProvider(StaticCredentialsProvider.create(
                AwsBasicCredentials.create(
                    localstack.getAccessKey(),
                    localstack.getSecretKey())))
            .build();
    }

    @Bean
    public SnsClient snsClient() {
        return SnsClient.builder()
            .region(Region.US_EAST_1)
            .endpointOverride(
                localstack.getEndpointOverride(LocalStackContainer.Service.SNS))
            .credentialsProvider(StaticCredentialsProvider.create(
                AwsBasicCredentials.create(
                    localstack.getAccessKey(),
                    localstack.getSecretKey())))
            .build();
    }
}