Slides | Coding Cloud

Blazing Fast SQS Listener

Tue, 11 Jun 2024 00:00:00 +0000

Blazing Fast

Cost Effective

& Resilient

AWS SQS Listener

Hi there, I’m Sanjay 👋!

About me

🔭 Working: Lead Software Engineer - working on Spring Boot, Reactive Programming, Microservices, Kafka, Cassandra, Kubernetes, AWS.
🖥️ Interests: I love building cool Software & Systems, Self-Hosting, Gaming
🌱 Learning: Go | Rust | Scala | Design Patterns
💬 Ask me about: Java | Reactive Spring | Containers | AWS
🧑‍🤝‍🧑 Collaboration: Looking to collaborate on several projects over here, check out my GitHub

Languages, Frameworks and Platforms

What is AWS SQS

SQS offers a Secure, Durable, and Highly Available Hosted Queue
Used to integrate and decouple distributed software systems and components
Standard queues support at-least-once message delivery, and FIFO queues support exactly-once message processing and high-throughput mode
Offers Dead Letter Queues

SQS Concepts w.r.t Lambda

Queue Types: Standard and FIFO
Visibility timeout: wait time after which message is visible again if not deleted after processing
In-Flight messages: messages that are received by the consumer but not deleted
SQS Event Source Mapping: Lambda Service reads items from a SQS & invokes Lambda Function
Enable trigger: Enable or disable SQS-Lambda Integration
Batch size: number of records to send to the function in each batch. Standard: 10,000(max). FIFO: 10(max)
Batch window: wait time (in second) to gather records before invoking the function, applicable for Standard Queue
Reserved Concurrency: the maximum number of concurrent instances allocated to your function.
Maximum Concurrency: the number of concurrent instances of the function that an Amazon SQS event source can invoke.

Message Lifecycle

SQS Visibility Timeout

To avoid message loss, it's consumers responsibility to delete the message after processing
Message remains in queue after it is received, but SQS sets a visibility timeout to prevent other consumer from processing same message again
Default visibility timeout is 30 seconds. Can be set between 0 seconds to 12 hours
Can be set a Queue level or dynamically changed per message
Caution: When using FIFO make sure to use Message GroupId which provides high distribution to avoid blocking processing due to error

Note:

For Standard queues, the visibility timeout isn't a guarantee against receiving a message twice.
FIFO queues allow the producer or consumer to attempt multiple retries: producers can retry send using deduplicationId and consumers doesn't receive messages for same message groupId unless deleted or timed-out

SQS Lambda Integration Architecture

FIVE Instances Lambda Service polls SQS queue every WaitTimeSeconds secs
(defaults to 20s). More info here.*
Function code is invoked when BatchSize number of messages are accumulated or MaximumBatchingWindowInSeconds are passed after receiving first message.
Function invocation is scaled if more messages are available in queue.
Function responds with List of failed messages (if ReportBatchItemFailures is enabled) or Success/Error.
Lambda Service deletes the successfully processed messages.

Note: *There is no clear documentation from AWS regarding SQS polling frequency.

Lambda Scaling

Standard Queues: Lambda uses long polling & reads up to 5 batches to invoke your function.
Up until 06 Nov 2023, the Lambda was adding up to 60 concurrent executions/minute, scaling up to a maximum of 1,250 concurrent executions in approximately 20 minutes.
Now Lambda functions can scale up to 5x faster, adding up to 300 concurrent executions/minute.
Even at peak performance, the Maximum number of messages processed concurrently by Lambda = 10 to 10K messages in a batch x 1250 executions = 12.5K to 12.5M
FIFO queues: Lambda sends messages to your function in the order that it receives them and ensures that messages in the same group are delivered to Lambda in order.
Lambda sorts the messages into groups and sends only one batch at a time for a group.
Your function can scale in concurrency to the number of active message groups.

Faster polling scale-up for AWS Lambda

How can we do this ourselves? But Better!

Implement a Dynamic SQS Poller, based on Number of Messages.
Convert Message to POJO & Concurrently Invoke Processing Logic.
For FIFO Queues, make sure to process messages from same GroupId in Sequence, while processing the Groups Concurrently.
Categorize Exceptions into Retryable & Non-Retryable.
Implement Error Handling, Retries and Timeout.
Publish the Failed Messages Directly to the DLQ and Mark it as Processed.
Collect the Processed messages & Delete those in Batch.
Implement Listener Health-check - when Listener stops polling/processing, set the health to DOWN.

Code

Questions?

AWS SQS + Lambda Integration

Thu, 20 Apr 2023 00:00:00 +0000

AWS SQS

Lambda Integration

with Resiliency Patterns

Documentation

Hi there, I’m Sanjay 👋!

About me

🔭 Working: Lead Software Engineer - working on Spring Boot, Reactive Programming, Microservices, Kafka, Cassandra, Kubernetes, AWS.
🖥️ Interests: I love building cool Software & Systems, Self-Hosting, Gaming
🌱 Learning: Go | Rust | Scala | Design Patterns
💬 Ask me about: Java | Reactive Spring | Containers | AWS
🧑‍🤝‍🧑 Collaboration: Looking to collaborate on several projects over here, check out my GitHub

Languages, Frameworks and Platforms

What is AWS SQS

SQS offers a Secure, Durable, and Highly Available Hosted Queue
Used to integrate and decouple distributed software systems and components
Standard queues support at-least-once message delivery, and FIFO queues support exactly-once message processing and high-throughput mode
Offers Dead Letter Queues

SQS Concepts w.r.t Lambda

Queue Types: Standard and FIFO
Visibility timeout: wait time after which message is visible again if not deleted after processing
Message retention period: 1 minute to 14 days (defaults to 4 days)
Delivery delay: 0 seconds to 15 minutes
SQS Event Source Mapping: Lambda Service reads items from a SQS & invokes Lambda Function
Enable trigger: Enable or disable SQS-Lambda Integration
Batch size: number of records to send to the function in each batch. Standard: 10,000(max). FIFO: 10(max)
Batch window: wait time (in second) to gather records before invoking the function, applicable for Standard

Message Lifecycle

SQS Visibility Timeout

To avoid message loss, it's consumers responsibility to delete the message after processing
Message remains in queue after it is received, but SQS sets a visibility timeout to prevent other consumer from processing same message again
Default visibility timeout is 30 seconds. Can be set between 0 seconds to 12 hours
Can be set a Queue level or dynamically changed per message
Caution: When using FIFO make sure to use Message GroupId which provides high distribution to avoid blocking processing due to error

Note:

For Standard queues, the visibility timeout isn't a guarantee against receiving a message twice.
FIFO queues allow the producer or consumer to attempt multiple retries: producers can retry send using deduplicationId and consumers doesn't receive messages for same message groupId unless deleted or timed-out

In-Flight Messages & Scaling

Messages that are received by the consumer but not deleted are known as In-Flight messages
For Standard queues: there can be a maximum of approximately 120,000 in flight messages
For FIFO queues: there can be a maximum of 20,000 in flight messages

Standard Queues: Lambda uses long polling & reads up to 5 batches to invoke your function.
Up until 06 Nov 2023, the Lambda was adding up to 60 concurrent executions/minute, scaling up to a maximum of 1,250 concurrent executions in approximately 20 minutes.
Now Lambda functions can scale up to 5x faster, adding up to 300 concurrent executions/minute.
Even at peak performance, the Maximum number of messages processed concurrently by Lambda = 10 messages in a batch * 1250 executions = 12,500
FIFO queues: Lambda sends messages to your function in the order that it receives them and ensures that messages in the same group are delivered to Lambda in order.
Lambda sorts the messages into groups and sends only one batch at a time for a group.
Your function can scale in concurrency to the number of active message groups.

SQS Message Event in Lambda Function - Standard

{
  "Records": [
    {
      "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
      "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
      "body": "Test message.",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1545082649183",
        "SenderId": "AIDAIENQZJOLO23YVJ4VO",
        "ApproximateFirstReceiveTimestamp": "1545082649185"
      },
      "messageAttributes": {},
      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
      "awsRegion": "us-east-2"
    },
    ...
  ]
}

SQS Message Event in Lambda Function - FIFO

{
  "Records": [
    {
      "messageId": "11d6ee51-4cc7-4302-9e22-7cd8afdaadf5",
      "receiptHandle": "AQEBBX8nesZEXmkhsmZeyIE8iQAMig7qw...",
      "body": "Test message.",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1573251510774",
        "SequenceNumber": "18849496460467696128",
        "MessageGroupId": "1",
        "SenderId": "AIDAIO23YVJENQZJOL4VO",
        "MessageDeduplicationId": "1",
        "ApproximateFirstReceiveTimestamp": "1573251510774"
      },
      "messageAttributes": {},
      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:fifo.fifo",
      "awsRegion": "us-east-2"
    }
  ]
}

SQS Resiliency Pattern with Lambda

Enable Batch Error Reporting by running:

aws lambda update-event-source-mapping \
    --uuid "a1b2c3d4-5678-90ab-cdef-123401" \
    --function-response-types "ReportBatchItemFailures"

Update your Lambda Function code to collect messageId of failed messages:

{
  "batchItemFailures": [
    {
      "itemIdentifier": "message-id-1"
    },
    {
      "itemIdentifier": "message-id-4"
    }
  ]
}

Blog Post on Batch Error Reporting by srcecde

Code

Questions?

AWS Lambda Deep Dive

Sun, 26 Mar 2023 00:00:00 +0000

AWS Lambda

Deep Dive

Documentation

Hi there, I’m Sanjay 👋!

About me

🔭 Working: Lead Software Engineer - working on Spring Boot, Reactive Programming, Microservices, Kafka, Cassandra, Kubernetes, AWS.
🖥️ Interests: I love building cool Software & Systems, Self-Hosting, Gaming
🌱 Learning: Go | Rust | Scala | Design Patterns
💬 Ask me about: Java | Reactive Spring | Containers | AWS
🧑‍🤝‍🧑 Collaboration: Looking to collaborate on several projects over here, check out my GitHub

Languages, Frameworks and Platforms

What is AWS Lambda Function

A Compute Service to run code without Provisioning or Managing Servers
Uses Containerization tech called Firecracker
Handles Server & OS Maintenance, Provisioning, Auto-Scaling, Logging
Code in one of the Languages supported by Lambda Runtimes

Lambda Concepts

Function: a resource invoked to run your code
Trigger: a resource/config that invokes a Lambda Function
Event: a JSON-formatted data for a Lambda Function to Process
Event Source Mapping: a resource that reads items from a Stream/Queue & invokes a function
Execution Env: provides a secure and isolated Runtime Environment for your Lambda function
Runtime: provides a lang-specific Environment that runs in an Execution Environment

What is Lambda Runtimes

Runtime provides a lang-specific env that runs in an execution env
It relays invocation Events, Context, & Responses between Lambda & the Function
Use AWS provided Lambda Runtimes or Build your own by implementing Runtime API

Execution Environment & Runtime

Supported Lambda Runtimes


Node.js	v18, v16, v14 & v12*
Python	v3.9, v3.8, v3.7
Java	v8, v11
C#	.NET v5, v6 & .NET Core v3.1
Go	v1.x
Ruby	v2.7*
Custom Runtime	- Based on Amazon Linux

Micronaut Custom Runtime Implementation Example

Lambda Function Java

public class MsgHandler implements RequestHandler<String, String> {
    public String handleRequest(String input, Context ctx) {
        context.getLogger().log("Input: " + input);
        return "Hello World - " + input;
    }
}

public class StreamHandler implements RequestStreamHandler {
    public void handleRequest(InputStream is, OutputStream os, Context ctx) {
        String input = IOUtils.toString(is, "UTF-8");
        os.write(("Hello World - " + input).getBytes());
    }
}

public class NoInterfaceHandler {
    public String customHandle(String input, Context context) {
        context.getLogger().log("Input: " + input);
        return "Hello World - " + input;
    }
}

Features

Automatically manages the underlying Compute Resources
Concurrency and Scaling controls
Pay Per Request Execution is millis
Multiple Event Sources: API-GW, SQS, S3, DynamoDB, etc.
Functions defined as Container Images
Lambda Extensions
Execute inside Private Subnets inside a VPC
File Systems Access + EFS

Limitations

Memory: 128 MB to 10,240 MB
No Direct Way to assign CPU: 1,769 MB memory is equivalent of one vCPU
Function Timeouts: 15 minutes
Invocation Payload: 6 MB (sync) & 256 KB (async)
File descriptors: 1024
Execution processes/threads: 1024
No Async Processing Allowed: Lambda Instance Freezes once invocation is completed
1 Lambda Instance serves only 1 request at time

Other Compute Options

Services	Comparison with Lambda
AWS EC2	Not Serverless, More RAM/CPU/Storage, Always ON
AWS EKS	Managed Container Platform on EC2
Fargate	Serverless Container Platform on EC2
Beanstalk	AWS Container based PaaS
GCP Cloud RUN	Based on OSS Knative - Charged for Runtime only