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:** ```xml <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>sqs</artifactId> </dependency> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>sns</artifactId> </dependency> ``` **Client Configuration:** ```java 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:** ```java 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:** ```java 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:** ```java 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:** ```java 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:** ```java 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 ```java @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:** ```java // 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 ```bash 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-sqs-operations.md) - [references/detailed-sns-operations.md](references/detailed-sns-operations.md) - [references/spring-boot-integration.md](references/spring-boot-integration.md) - [references/aws-official-documentation.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