Amazon SQS #

Amazon Simple Queue Service (SQS) is AWS’s fully managed message queuing service — you don’t need to manage servers, configuration, or scaling. SQS is well suited to AWS-based architectures where components need to communicate asynchronously without tight coupling. It comes in two variants: Standard Queues (nearly unlimited throughput, at-least-once delivery, no ordering guarantees) and FIFO Queues (exactly-once delivery, guaranteed ordering, but more limited throughput). In Kotlin, you use the AWS SDK for Java v2 which supports coroutines through its Netty-based async client. This article covers configuration, message CRUD operations, efficient consumption patterns, SNS integration, and how to develop with LocalStack without AWS costs.

SQS Standard vs FIFO #

flowchart TD
    A{Main Need?} --> B{Is message\nordering important?}
    B -- Yes --> C{Throughput\n> 3000/sec?}
    C -- Yes --> D["Standard Queue\n+ order in the consumer\nor use Kafka"]
    C -- No --> E["FIFO Queue\nMax 3000 msg/s\nExact-once delivery"]
    B -- No --> F{Very large\nmessage volume?}
    F -- Yes --> G["Standard Queue\nUnlimited throughput\nAt-least-once"]
    F -- No --> G

Aspect	Standard Queue	FIFO Queue
Ordering	Best-effort (not guaranteed)	Guaranteed (FIFO)
Delivery	At-least-once (can duplicate)	Exactly-once
Throughput	Nearly unlimited	3,000 msg/sec (batch), 300 msg/sec
Name	`queue-name`	`queue-name.fifo` (must end with `.fifo`)
Deduplication	None	Automatic with Deduplication ID
Price	Cheaper	Slightly more expensive
Best for	Logs, notifications, general task queues	Financial transactions, orders

Setup and Dependencies #

// build.gradle.kts
dependencies {
    // AWS SDK v2 — SQS
    implementation("software.amazon.awssdk:sqs:2.25.27")

    // AWS SDK v2 — STS (for assume role, optional)
    implementation("software.amazon.awssdk:sts:2.25.27")

    // HTTP client for SDK v2 (choose one)
    implementation("software.amazon.awssdk:netty-nio-client:2.25.27")  // async
    // or:
    // implementation("software.amazon.awssdk:apache-client:2.25.27")  // sync

    // kotlinx.serialization
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")

    // Coroutines
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
}

Creating the SQS Client #

import software.amazon.awssdk.auth.credentials.*
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.sqs.SqsClient
import software.amazon.awssdk.services.sqs.SqsAsyncClient
import java.net.URI

object SqsConnection {

    // Synchronous client (blocking)
    val client: SqsClient by lazy {
        SqsClient.builder()
            .region(Region.AP_SOUTHEAST_1)  // Asia Pacific (Singapore)
            .credentialsProvider(
                DefaultCredentialsProvider.create()
                // The SDK looks for credentials in:
                // 1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
                // 2. ~/.aws/credentials
                // 3. EC2 Instance Profile / ECS Task Role
                // 4. IAM Role for Service Account (EKS)
            )
            .build()
    }

    // Client for LocalStack (local development without AWS)
    fun localClient(endpoint: String = "http://localhost:4566"): SqsClient {
        return SqsClient.builder()
            .region(Region.AP_SOUTHEAST_1)
            .endpointOverride(URI.create(endpoint))
            .credentialsProvider(
                StaticCredentialsProvider.create(
                    AwsBasicCredentials.create("test", "test")  // LocalStack accepts any key
                )
            )
            .build()
    }

    // Async client with coroutines
    val asyncClient: SqsAsyncClient by lazy {
        SqsAsyncClient.builder()
            .region(Region.AP_SOUTHEAST_1)
            .credentialsProvider(DefaultCredentialsProvider.create())
            .build()
    }
}

Creating and Managing Queues #

import software.amazon.awssdk.services.sqs.model.*

fun createQueue(name: String, isFifo: Boolean = false): String {
    val client = SqsConnection.client
    val finalName = if (isFifo && !name.endsWith(".fifo")) "$name.fifo" else name

    val attributes = mutableMapOf<QueueAttributeName, String>()

    if (isFifo) {
        attributes[QueueAttributeName.FIFO_QUEUE] = "true"
        attributes[QueueAttributeName.CONTENT_BASED_DEDUPLICATION] = "true"
    }

    // Common configuration
    attributes[QueueAttributeName.VISIBILITY_TIMEOUT] = "30"       // 30 seconds
    attributes[QueueAttributeName.MESSAGE_RETENTION_PERIOD] = "345600"  // 4 days
    attributes[QueueAttributeName.RECEIVE_MESSAGE_WAIT_TIME_SECONDS] = "20"  // long polling

    val response = client.createQueue { req ->
        req.queueName(finalName).attributes(attributes)
    }

    println("Queue created: ${response.queueUrl()}")
    return response.queueUrl()
}

fun getQueueUrl(name: String): String {
    return SqsConnection.client.getQueueUrl { req ->
        req.queueName(name)
    }.queueUrl()
}

fun queueAttributes(queueUrl: String): Map<QueueAttributeName, String> {
    return SqsConnection.client.getQueueAttributes { req ->
        req.queueUrl(queueUrl)
           .attributeNames(QueueAttributeName.ALL)
    }.attributes()
}

fun deleteQueue(queueUrl: String) {
    SqsConnection.client.deleteQueue { req -> req.queueUrl(queueUrl) }
    println("Queue deleted: $queueUrl")
}

Sending Messages #

import kotlinx.serialization.Serializable
import kotlinx.serialization.json.Json

@Serializable
data class OrderEvent(
    val orderId: String,
    val userId: Long,
    val total: Double,
    val status: String,
    val timestamp: Long = System.currentTimeMillis()
)

class SqsProducer(private val queueUrl: String) {
    private val client = SqsConnection.client
    private val json = Json { ignoreUnknownKeys = true }

    // Send a single message
    fun send(event: OrderEvent): String {
        val payload = json.encodeToString(OrderEvent.serializer(), event)

        val request = SendMessageRequest.builder()
            .queueUrl(queueUrl)
            .messageBody(payload)
            .delaySeconds(0)  // send immediately (can be delayed 0-900 seconds)
            .messageAttributes(mapOf(
                "tipe" to MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue("OrderEvent")
                    .build(),
                "versi" to MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue("1.0")
                    .build()
            ))
            .build()

        val response = client.sendMessage(request)
        println("Message sent — ID: ${response.messageId()}, MD5: ${response.md5OfMessageBody()}")
        return response.messageId()
    }

    // Send to a FIFO queue (needs MessageGroupId and an optional DeduplicationId)
    fun sendToFifo(event: OrderEvent): String {
        val payload = json.encodeToString(OrderEvent.serializer(), event)

        val request = SendMessageRequest.builder()
            .queueUrl(queueUrl)
            .messageBody(payload)
            .messageGroupId(event.userId.toString())  // ordered messages per user
            .messageDeduplicationId(event.orderId)    // avoid duplicates (if not using content-based)
            .build()

        return client.sendMessage(request).messageId()
    }

    // Send many messages at once — batch (max 10 messages per batch)
    fun sendBatch(events: List<OrderEvent>): Pair<Int, Int> {
        var success = 0
        var failed = 0

        events.chunked(10).forEach { batch ->
            val entries = batch.mapIndexed { idx, event ->
                SendMessageBatchRequestEntry.builder()
                    .id("$idx")  // unique ID within the batch, not the message ID
                    .messageBody(json.encodeToString(OrderEvent.serializer(), event))
                    .build()
            }

            val response = client.sendMessageBatch { req ->
                req.queueUrl(queueUrl).entries(entries)
            }

            success += response.successful().size
            failed += response.failed().size

            if (response.failed().isNotEmpty()) {
                response.failed().forEach { f ->
                    println("Failed to send entry ${f.id()}: ${f.message()}")
                }
            }
        }

        println("Batch done — success: $success, failed: $failed")
        return Pair(success, failed)
    }
}

Receiving and Deleting Messages #

class SqsConsumer(private val queueUrl: String) {
    private val client = SqsConnection.client
    private val json = Json { ignoreUnknownKeys = true }

    // Receive messages once (for testing or manual polling)
    fun receiveMessages(max: Int = 10): List<Message> {
        val request = ReceiveMessageRequest.builder()
            .queueUrl(queueUrl)
            .maxNumberOfMessages(max)  // SQS max 10 per request
            .waitTimeSeconds(20)         // long polling — wait up to 20 seconds if the queue is empty
            .visibilityTimeout(30)       // hide messages 30 seconds from other consumers
            .messageAttributeNames("All")
            .build()

        return client.receiveMessage(request).messages()
    }

    // Delete a message after successful processing
    fun deleteMessage(receiptHandle: String) {
        client.deleteMessage { req ->
            req.queueUrl(queueUrl).receiptHandle(receiptHandle)
        }
    }

    // Delete many messages at once
    fun deleteBatch(messages: List<Message>) {
        messages.chunked(10).forEach { batch ->
            val entries = batch.mapIndexed { idx, msg ->
                DeleteMessageBatchRequestEntry.builder()
                    .id("$idx")
                    .receiptHandle(msg.receiptHandle())
                    .build()
            }
            client.deleteMessageBatch { req ->
                req.queueUrl(queueUrl).entries(entries)
            }
        }
    }

    // Return a message to the queue sooner (if it can't be processed right now)
    fun returnMessage(receiptHandle: String, newVisibilityTimeout: Int = 0) {
        client.changeMessageVisibility { req ->
            req.queueUrl(queueUrl)
               .receiptHandle(receiptHandle)
               .visibilityTimeout(newVisibilityTimeout)  // 0 = visible again immediately
        }
    }

    // Consumer loop — poll continuously
    fun startLoop(running: () -> Boolean = { true }) {
        println("SQS consumer started polling: $queueUrl")

        while (running()) {
            val messages = receiveMessages(max = 10)

            if (messages.isEmpty()) {
                // Long polling already waited 20 seconds, still empty — normal
                continue
            }

            val successfullyDeleted = mutableListOf<Message>()

            messages.forEach { msg ->
                try {
                    processMessage(msg)
                    successfullyDeleted.add(msg)
                } catch (e: Exception) {
                    println("Failed to process ${msg.messageId()}: ${e.message}")
                    // Don't delete — let the visibility timeout expire
                    // → the message will become visible again and can be retried
                    // → after maxReceiveCount failures → it goes to the Dead Letter Queue
                }
            }

            // Delete only what was processed successfully
            if (successfullyDeleted.isNotEmpty()) {
                deleteBatch(successfullyDeleted)
                println("${successfullyDeleted.size} messages processed and deleted")
            }
        }
    }

    private fun processMessage(msg: Message) {
        val event = json.decodeFromString(OrderEvent.serializer(), msg.body())
        val type = msg.messageAttributes()["tipe"]?.stringValue()

        println("Processing message ${msg.messageId()}: order ${event.orderId} " +
                "(type: $type, received: ${msg.attributes()[MessageSystemAttributeName.APPROXIMATE_RECEIVE_COUNT]} times)")
    }
}

Dead Letter Queues (DLQ) #

A DLQ receives messages that failed processing after a number of attempts (maxReceiveCount):

fun setupQueueWithDlq(mainName: String, dlqName: String): Pair<String, String> {
    val client = SqsConnection.client

    // 1. Create the DLQ first
    val dlqUrl = createQueue(dlqName)

    // Get the DLQ's ARN
    val dlqArn = client.getQueueAttributes { req ->
        req.queueUrl(dlqUrl).attributeNames(QueueAttributeName.QUEUE_ARN)
    }.attributes()[QueueAttributeName.QUEUE_ARN]!!

    // 2. Create the main queue with a Redrive Policy
    val redrivePolicy = """
        {
            "maxReceiveCount": "3",
            "deadLetterTargetArn": "$dlqArn"
        }
    """.trimIndent()

    val mainUrl = client.createQueue { req ->
        req.queueName(mainName)
           .attributes(mapOf(
               QueueAttributeName.REDRIVE_POLICY to redrivePolicy,
               QueueAttributeName.VISIBILITY_TIMEOUT to "30",
               QueueAttributeName.RECEIVE_MESSAGE_WAIT_TIME_SECONDS to "20"
           ))
    }.queueUrl()

    println("Main queue: $mainUrl")
    println("Dead Letter Queue: $dlqUrl")
    return Pair(mainUrl, dlqUrl)
}

// Monitor the DLQ — process failed messages manually
fun monitorDlq(dlqUrl: String) {
    val consumer = SqsConsumer(dlqUrl)
    val messages = consumer.receiveMessages(max = 10)

    println("Messages in the DLQ: ${messages.size}")
    messages.forEach { msg ->
        val receiveCount = msg.attributes()[MessageSystemAttributeName.APPROXIMATE_RECEIVE_COUNT]
        println("  MessageId: ${msg.messageId()}, received: $receiveCount times")
        println("  Body: ${msg.body().take(100)}...")
    }
}

Consumers with Coroutines #

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import software.amazon.awssdk.services.sqs.model.*

fun sqsFlow(
    queueUrl: String,
    scope: CoroutineScope
): Flow<Message> = flow {
    val client = SqsConnection.client

    while (scope.isActive) {
        val messages = withContext(Dispatchers.IO) {
            client.receiveMessage { req ->
                req.queueUrl(queueUrl)
                   .maxNumberOfMessages(10)
                   .waitTimeSeconds(20)
            }.messages()
        }

        messages.forEach { emit(it) }
    }
}

fun main() = runBlocking {
    val queueUrl = getQueueUrl("pesanan-events")
    val json = Json { ignoreUnknownKeys = true }
    val client = SqsConnection.client

    sqsFlow(queueUrl, this)
        .map { msg ->
            val event = json.decodeFromString(OrderEvent.serializer(), msg.body())
            Pair(msg, event)
        }
        .filter { (_, event) -> event.status == "NEW" }
        .collect { (msg, event) ->
            println("Processing new order: ${event.orderId}")

            // Delete after successful processing
            withContext(Dispatchers.IO) {
                client.deleteMessage { req ->
                    req.queueUrl(queueUrl).receiptHandle(msg.receiptHandle())
                }
            }
        }
}

SNS + SQS Integration (Fan-out Pattern) #

SNS (Simple Notification Service) combined with SQS sends one message to many queues at once:

import software.amazon.awssdk.services.sns.SnsClient
import software.amazon.awssdk.services.sns.model.*

fun setupSnsSqsFanout() {
    val sqsClient = SqsConnection.client
    val snsClient = SnsClient.builder().region(Region.AP_SOUTHEAST_1).build()

    // 1. Create the SNS topic
    val topicArn = snsClient.createTopic { req ->
        req.name("pesanan-events")
    }.topicArn()

    // 2. Create several SQS queues
    val emailUrl = createQueue("notif-email")
    val auditUrl = createQueue("audit-log")
    val inventoryUrl = createQueue("update-inventori")

    // 3. Subscribe each queue to the SNS topic
    listOf(emailUrl, auditUrl, inventoryUrl).forEach { queueUrl ->
        val queueArn = sqsClient.getQueueAttributes { req ->
            req.queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN)
        }.attributes()[QueueAttributeName.QUEUE_ARN]!!

        snsClient.subscribe { req ->
            req.topicArn(topicArn)
               .protocol("sqs")
               .endpoint(queueArn)
        }

        println("Queue $queueUrl subscribed to SNS $topicArn")
    }

    // 4. Now publish to SNS → automatically goes to all queues
    snsClient.publish { req ->
        req.topicArn(topicArn)
           .message("""{"orderId":"ORD-001","status":"NEW"}""")
           .subject("NewOrder")
    }
    println("Message fanned out to all queues via SNS")
}

Development with LocalStack #

LocalStack lets you run AWS services locally using Docker, at no cost:

# docker-compose.yml
services:
  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=sqs,sns,s3
      - DEFAULT_REGION=ap-southeast-1

# Run LocalStack
docker compose up -d

# Create a queue via the AWS CLI (point it at LocalStack)
aws --endpoint-url=http://localhost:4566 \
    --region ap-southeast-1 \
    sqs create-queue --queue-name pesanan-events

# List queues
aws --endpoint-url=http://localhost:4566 \
    --region ap-southeast-1 \
    sqs list-queues

// In Kotlin code, use a client with an endpoint override
val localClient = SqsConnection.localClient("http://localhost:4566")
val queueUrl = localClient.createQueue { req -> req.queueName("pesanan-events") }.queueUrl()
println("Local queue: $queueUrl")
// Output: http://localhost:4566/000000000000/pesanan-events

Cost and Performance Tips #

// 1. Always use Long Polling (waitTimeSeconds = 20)
//    Short polling: billed per API request, including empty requests
//    Long polling: waits up to 20 seconds before returning empty
//    Save up to 80% on API call costs for rarely-filled queues

// 2. Batch as much as possible
//    sendMessageBatch: max 10 messages in one API call
//    receiveMessage: max 10 messages in one API call
//    deleteMessageBatch: max 10 messages in one API call

// 3. The visibility timeout must be longer than the processing time
//    If processing takes 5 minutes, set the visibility timeout to 6-7 minutes
//    If too short → messages reappear before processing finishes

// 4. Watch the message size
//    SQS max 256KB per message
//    For large payloads → store in S3, send the S3 reference in SQS

// 5. Monitoring via CloudWatch
//    Important metrics: ApproximateNumberOfMessagesVisible (queue backlog)
//                       ApproximateAgeOfOldestMessage (oldest message)
//                       NumberOfMessagesSent, NumberOfMessagesDeleted

Summary #

Standard Queues for high volume, FIFO for ordering — Standard Queues offer nearly unlimited throughput but don’t guarantee ordering and can produce duplicates. FIFO Queues guarantee ordering and exactly-once delivery but are limited to 3,000 messages/second.
Long polling is mandatory — set waitTimeSeconds=20 on every receiveMessage. This avoids thousands of wasteful API calls to empty queues and saves significant costs.
Delete messages only after successful processing — don’t delete messages when received. Delete only after processing succeeds. Failed messages automatically become visible again after the visibility timeout expires.
A Dead Letter Queue must be configured — set maxReceiveCount (usually 3-5) and point to a DLQ. This prevents “poison messages” from blocking the queue forever.
The visibility timeout must be longer than the processing time — if processing can take a long time, set a longer visibility timeout or extend it dynamically with changeMessageVisibility.
Batch for cost efficiency — use sendMessageBatch, deleteMessageBatch to reduce the number of API calls. SQS is billed per API call, not per message.
SNS + SQS for fan-out — to send one event to many independent consumers, use SNS in front of several SQS queues. Cleaner than sending to many queues separately from the producer.
LocalStack for development — use LocalStack with Docker for free local testing. The endpoint override in the SDK is enough to point all calls at LocalStack.

← Previous: RabbitMQ Next: Google Pub/Sub →