AWS Lambda

Lambda Event Sources Guide

Overview

This guide covers the most common Lambda event sources — both polling-based (Event Source Mappings) and push-based (S3 notifications, SNS subscriptions). Use esm_guidance for polling source setup recommendations and esm_optimize for performance tuning.

Delivery guarantee: All Lambda event sources deliver events at least once. Your function must be idempotent — the same record may be processed more than once. Use the AWS Lambda Powertools Idempotency utility (backed by DynamoDB) to handle duplicates safely.

Polling-Based Event Sources (ESM)

Event Source Mappings use a Lambda-managed poller that reads from the source, batches records, and invokes your function. You control throughput via batch size, concurrency, and parallelization.

DynamoDB Streams

Use case: React to data changes in DynamoDB tables

Key configuration:

• StartingPosition: LATEST for new records only, TRIM_HORIZON for all

• BatchSize: 1-10000 (default 100)

• ParallelizationFactor: 1-10 (default 1, increase for throughput)

• BisectBatchOnFunctionError: Enable to isolate poison records

• MaximumRetryAttempts: Set to prevent infinite retries (default unlimited)

• MaximumBatchingWindowInSeconds: Buffer time before invoking (0-300)

Best practices:

• Enable BisectBatchOnFunctionError and set MaximumRetryAttempts to 3

• Configure a dead-letter queue for records that exhaust retries

• Use ParallelizationFactor > 1 when processing can't keep up

Kinesis Streams

Use case: Process real-time streaming data

Key configuration:

• BatchSize: 1-10000 (default 100)

• ParallelizationFactor: 1-10 (should not exceed shard count)

• MaximumBatchingWindowInSeconds: Buffer time (0-300)

• TumblingWindowInSeconds: For aggregation scenarios (0-900)

• StartingPosition: LATEST or TRIM_HORIZON

Best practices:

• Higher batch sizes reduce invocation costs but increase timeout risk

• Use tumbling windows for time-based aggregation (counts, sums, averages)

• Enable enhanced fan-out when multiple consumers read from the same stream

SQS Queues

Use case: Decouple components with reliable messaging

Key configuration:

• BatchSize: 1-10000 (default 10)

• MaximumBatchingWindowInSeconds: Buffer time (0-300)

• MaximumConcurrency: Limit concurrent Lambda invocations

• FunctionResponseTypes: Set to ["ReportBatchItemFailures"] to avoid reprocessing successful messages

FIFO queue considerations:

• Use BatchSize: 1 for strict ordering

• Limit MaximumConcurrency to prevent out-of-order processing

• Use message group IDs for parallel processing within groups

Python SQS batch processor with partial failure reporting:

from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.batch import (
    BatchProcessor, EventType, process_partial_response,
)
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

logger = Logger()
processor = BatchProcessor(event_type=EventType.SQS)

def process_record(record: SQSRecord):
    body = record.json_body
    logger.info("Processing order", order_id=body["orderId"])
    # Business logic here — raise to mark this record as failed
    return {"orderId": body["orderId"], "status": "processed"}

@logger.inject_lambda_context
def handler(event, context):
    return process_partial_response(
        event=event, record_handler=process_record, processor=processor, context=context,
    )

TypeScript SQS batch processor:

import {
  BatchProcessor,
  EventType,
  processPartialResponse,
} from '@aws-lambda-powertools/batch';
import { Logger } from '@aws-lambda-powertools/logger';
import type { SQSHandler, SQSRecord } from 'aws-lambda';

const processor = new BatchProcessor(EventType.SQS);
const logger = new Logger();

const recordHandler = async (record: SQSRecord): Promise<void> => {
  const body = JSON.parse(record.body);
  logger.info('Processing order', { orderId: body.orderId });
};

export const handler: SQSHandler = async (event, context) =>
  processPartialResponse(event, recordHandler, processor, { context });

Both examples require FunctionResponseTypes: ["ReportBatchItemFailures"] in the ESM configuration.

Best practices:

• Always enable ReportBatchItemFailures for partial failure handling

• Set queue VisibilityTimeout >= Lambda function timeout

• Configure a DLQ with maxReceiveCount of 3-5

MSK/Kafka

Use case: Process high-throughput streaming data from Kafka

Key configuration:

• Topics: List of Kafka topics to consume

• BatchSize: 1-10000 (default 100)

• MaximumBatchingWindowInSeconds: Buffer time (0-300)

• StartingPosition: LATEST or TRIM_HORIZON

• ConsumerGroupId: Consumer group identifier

Network requirements:

• Lambda must have VPC access to the MSK cluster

• Security groups must allow traffic on ports 9092 (plaintext) or 9094 (TLS)

• Use IAM authentication or SASL/SCRAM for authentication

Best practices:

• Use esm_kafka_troubleshoot for connectivity issues

• Generate IAM policies with secure_esm_msk_policy

Powertools Kafka Consumer: For Kafka events with Avro, Protobuf, or JSON Schema payloads, use the Kafka Consumer utility to deserialize records automatically instead of manually parsing byte arrays:

from aws_lambda_powertools.utilities.kafka import KafkaConsumer, SchemaConfig

schema_config = SchemaConfig(schema_type="AVRO", schema_registry_url="https://my-registry.example.com")
consumer = KafkaConsumer(schema_config=schema_config)

@consumer.handler
def handler(event, context):
    for record in consumer.records:
        # record.value is already deserialized from Avro
        order = record.value
        print(f"Order: {order['orderId']}")

Available for Python, TypeScript, Java, and .NET. Supports Avro, Protobuf, and JSON Schema with both AWS Glue Schema Registry and Confluent Schema Registry.

Amazon MQ

Use case: Process messages from ActiveMQ or RabbitMQ brokers

Lambda connects to Amazon MQ using a VPC-attached ESM. Configure the broker in a private subnet and ensure Lambda's security group can reach the broker port (61614 for ActiveMQ over STOMP/TLS, 5671 for RabbitMQ over AMQP/TLS).

Self-Managed Kafka

Use case: Process messages from a Kafka cluster you operate (not MSK)

Use a self-managed Kafka ESM when your cluster is not AWS-managed. Lambda connects via the network configuration you provide. Supports SASL/PLAIN, SASL/SCRAM, mTLS, and VPC connectivity.

Amazon DocumentDB (with change streams)

Use case: React to changes in DocumentDB collections

Similar to DynamoDB Streams — Lambda polls the DocumentDB change stream. Requires DocumentDB change streams to be enabled on the collection and Lambda to have VPC access to the cluster.

Push-Based Event Sources

These sources invoke Lambda directly via asynchronous invocation — no ESM poller involved. Lower latency than polling sources, but less control over throughput and concurrency.

S3 Event Notifications

Use case: React to object uploads, deletions, or modifications — image processing, file validation, data import, thumbnail generation

SAM template:

ImageProcessor:
  Type: AWS::Serverless::Function
  Properties:
    Handler: src/handlers/process_image.handler
    Runtime: python3.12
    Architectures: [arm64]
    Policies:
      - S3ReadPolicy:
          BucketName: !Ref UploadBucket
    Events:
      ImageUploaded:
        Type: S3
        Properties:
          Bucket: !Ref UploadBucket
          Events: s3:ObjectCreated:*
          Filter:
            S3Key:
              Rules:
                - Name: prefix
                  Value: uploads/
                - Name: suffix
                  Value: .jpg

UploadBucket:
  Type: AWS::S3::Bucket

Key configuration:

• Events: Event types to trigger on — s3:ObjectCreated:*, s3:ObjectRemoved:*, s3:ObjectCreated:Put, etc.

• Filter.S3Key.Rules: Prefix and/or suffix filters to limit which objects trigger the function

• Bucket: Must reference an AWS::S3::Bucket declared in the same SAM template

Best practices:

• Avoid recursive triggers — if your function writes back to the same bucket that triggers it, Lambda will loop infinitely. Use a separate output bucket or a different prefix

• S3 delivers events at least once and NOT in order — write idempotent handlers and don't depend on event sequencing

• URL-decode the object key (urllib.parse.unquote_plus in Python, decodeURIComponent in JS) — S3 URL-encodes special characters in keys

• Use prefix/suffix filters to limit invocations to relevant objects

• For complex routing (multiple consumers, content-based filtering, cross-account), use S3 → EventBridge instead — enable EventBridge notifications on the bucket and create rules

Python S3 event handler:

import json
import urllib.parse
import boto3
from aws_lambda_powertools import Logger

logger = Logger()
s3 = boto3.client('s3')

@logger.inject_lambda_context
def handler(event, context):
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = urllib.parse.unquote_plus(record['s3']['object']['key'])
        size = record['s3']['object']['size']
        logger.info("Processing object", bucket=bucket, key=key, size=size)

        response = s3.get_object(Bucket=bucket, Key=key)
        # Process the object content

TypeScript S3 event handler:

import { Logger } from '@aws-lambda-powertools/logger';
import { Tracer } from '@aws-lambda-powertools/tracer';
import type { S3Event, Context } from 'aws-lambda';
import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3';

const logger = new Logger();
const tracer = new Tracer();
const s3 = tracer.captureAWSv3Client(new S3Client({}));

export const handler = async (event: S3Event, context: Context): Promise<void> => {
  logger.addContext(context);
  for (const record of event.Records) {
    const bucket = record.s3.bucket.name;
    const key = decodeURIComponent(record.s3.object.key.replace(/\+/g, ' '));
    const size = record.s3.object.size;
    logger.info('Processing object', { bucket, key, size });

    const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
    // Process the object content
  }
};

SNS Subscriptions

Use case: Fan-out processing — one published message triggers multiple independent Lambda consumers

SAM template:

OrderNotifier:
  Type: AWS::Serverless::Function
  Properties:
    Handler: src/handlers/notify.handler
    Runtime: python3.12
    Architectures: [arm64]
    Events:
      OrderEvent:
        Type: SNS
        Properties:
          Topic: !Ref OrderTopic
          FilterPolicy:
            event_type:
              - order_placed
              - order_shipped
          FilterPolicyScope: MessageAttributes
          RedrivePolicy:
            deadLetterTargetArn: !GetAtt OrderDLQ.Arn

OrderTopic:
  Type: AWS::SNS::Topic

OrderDLQ:
  Type: AWS::SQS::Queue
  Properties:
    MessageRetentionPeriod: 1209600  # 14 days

Key configuration:

• Topic: ARN or reference to the SNS topic

• FilterPolicy: JSON filter to receive only matching messages — reduces invocations and cost

• FilterPolicyScope: MessageAttributes (default) or MessageBody

• RedrivePolicy: DLQ ARN for messages that fail delivery — configured at the subscription level, not the topic

Best practices:

• Use filter policies to reduce invocations — filter at the subscription level, not in handler code

• Configure a redrive policy (DLQ) on every subscription — SNS retries server-side errors up to 100,015 times over 23 days; client-side errors (deleted function) go directly to DLQ

• Set DLQ MessageRetentionPeriod to 14 days (maximum) for investigation time

• SNS delivers at least once; write idempotent handlers

• FIFO SNS topics do NOT support Lambda subscriptions — use standard topics only

• For simple point-to-point delivery, prefer SQS → Lambda (ESM) over SNS → Lambda

• For complex event routing with pattern matching, prefer EventBridge over SNS

Python SNS event handler:

import json
from aws_lambda_powertools import Logger

logger = Logger()

@logger.inject_lambda_context
def handler(event, context):
    for record in event['Records']:
        message = json.loads(record['Sns']['Message'])
        subject = record['Sns'].get('Subject', '')
        message_id = record['Sns']['MessageId']
        logger.info("Processing SNS message", message_id=message_id, subject=subject)

        # Process the message

TypeScript SNS event handler:

import { Logger } from '@aws-lambda-powertools/logger';
import type { SNSEvent, Context } from 'aws-lambda';

const logger = new Logger();

export const handler = async (event: SNSEvent, context: Context): Promise<void> => {
  logger.addContext(context);
  for (const record of event.Records) {
    const message = JSON.parse(record.Sns.Message);
    const subject = record.Sns.Subject ?? '';
    const messageId = record.Sns.MessageId;
    logger.info('Processing SNS message', { messageId, subject });

    // Process the message
  }
};

Event Filtering

ESM Filtering (Polling Sources)

ESM event filtering lets Lambda evaluate filter criteria before invoking your function, reducing unnecessary invocations and costs.

Add filters in SAM template:

MyFunction:
  Type: AWS::Serverless::Function
  Events:
    MySQSEvent:
      Type: SQS
      Properties:
        Queue: !GetAtt MyQueue.Arn
        FilterCriteria:
          Filters:
            - Pattern: '{"body": {"eventType": ["ORDER_CREATED"]}}'

Filter pattern syntax matches against the event structure:

• Exact match: {"body": {"status": ["ACTIVE"]}}

• Prefix match: {"body": {"id": [{"prefix": "order-"}]}}

• Numeric range: {"body": {"amount": [{"numeric": [">", 100]}]}}

• Exists check: {"body": {"metadata": [{"exists": true}]}}

Filtering is supported for SQS, Kinesis, DynamoDB Streams, MSK, self-managed Kafka, and MQ. Records that don't match filters are dropped before Lambda is invoked — SQS messages are deleted, stream records are skipped.

Push Source Filtering

Push-based sources use their own filtering mechanisms (not ESM FilterCriteria):

• S3: Prefix/suffix filters on the object key via Filter.S3Key.Rules in the SAM template

• SNS: Subscription filter policies via FilterPolicy — supports matching on MessageAttributes (default) or MessageBody

ESM Provisioned Mode

For high-throughput Kafka (MSK or self-managed) and SQS workloads, provisioned mode provides:

• 3x faster autoscaling compared to default mode

• 16x higher maximum capacity

• Manual control over minimum and maximum event pollers

Enable in SAM template:

MySQSEvent:
  Type: SQS
  Properties:
    Queue: !GetAtt MyQueue.Arn
    ProvisionedPollerConfig:
      MinimumPollers: 2
      MaximumPollers: 20

Use provisioned mode when default autoscaling is too slow to absorb traffic spikes without message backlog.

Batch Size Guidelines

Error Handling

• Stream sources (DynamoDB, Kinesis): Records retry until success, expiry, or max retries. Enable BisectBatchOnFunctionError and set MaximumRetryAttempts.

• SQS: Failed messages return to the queue after visibility timeout. Use ReportBatchItemFailures for partial batch success.

• Kafka: Similar to stream sources. Failed batches retry based on ESM configuration.

Always configure a dead-letter queue or on-failure destination to capture records that cannot be processed.

Monitoring

For event source metrics to alarm on (IteratorAge, Errors, Throttles, DLQ depth), alarm thresholds, and dashboard setup, see observability.md.

Schema Integration

For type-safe event processing with EventBridge:

1. Use search_schema to find event schemas

2. Use describe_schema to get the full definition

3. Generate typed handlers based on the schema