How can I prevent an Amazon SQS message from invoking my Lambda function more than once?

Short description

Lambda supports at-least-once message delivery. In specific cases, the retry mechanism might send duplicates of the same message. Amazon SQS then sends the messages to your dead-letter queue, if you've configured one.

Duplicate Lambda invokes for the same Amazon SQS message can happen for one of the following reasons:

• Your function returns an error or times out.
• The Lambda service successfully processed the batch but failed to delete the message from the Amazon SQS queue before the visibility timeout expired.
• The Lambda service sent the event to the function but didn't receive an acknowledgement from the function.
• An intermittent issue caused Amazon SQS to return the same message and it's polled again by the Lambda service.
• The sum of the batch window and the function duration is longer than your Amazon SQS queue visibility timeout. The SQS visibility timeout must be at least six times the total of the function timeout and the batch window timeout.

To confirm that you received the same message multiple times, check the message ID. If you receive multiple copies of the same message, then the messages have the same ID.
Note: If you receive multiple messages with the same content but different message IDs, then the queue received the message more than once. Confirm that the producer runs, sends the messages to the correct queue URL, and has the required permissions.

Resolution

To resolve this issue, turn on report batch item failure in your Lambda function SQS trigger. Then, create a modular function code that iterates through the batch, processes, and deletes successful and duplicate messages. The function stores the messageID of successful messages in a Amazon DynamoDB table and then verifies that it processed the message earlier.

Important: The following resolution slows processing time by sending multiple requests to DynamoDB for each incoming message and results in higher costs from API calls.

Create a DynamoDB table

The following DynamoDB table holds your message IDs so that a Lambda function can compare them for duplication.

Complete the following steps:

1. Open the DynamoDB console.
2. Choose Create table.
3. In the Create DynamoDB table page, enter the following values:
   For Table, enter ProcessedRecords
   Under Partition key, enter Records
   Set the data type to String
4. Choose Create table.

Create a Lambda function

Important: The Lambda function code must be idempotent. For example function idempotent logic, see Best practices for working with AWS Lambda functions.

Create a Lambda function that compares incoming messages with messages that were previously successful and then held in your DynamoDB table. If a message was previously successful, then the function doesn't allow duplicates to process. If new, unique messages are successful, then they're sent to the table for future comparison.

SQS retries failed messages until Lambda successfully processes them or until the ReceiveCount for a message exceeds the maxReceiveCount.

In your function, add an execution role to allow the dynamodb:Query and dynamodb:PutItem actions. Then, report batch item failures in the SQS trigger to identify and skip duplicate messages in the batch.

The following example Lambda function logic converts the message body to upper case:

import boto3

dynamodb_client = boto3.client('dynamodb')

DDB_TABLE = 'ProcessedRecords'

# Validates if the message is already processed in previous invokes.
# @input string message_id
#
# @param message_id used to query the message from DynamoDB
# @return Boolean
def is_duplicate_message(message_id):
    return dynamodb_client.query(
        TableName = DDB_TABLE
        Select='COUNT',
        KeyConditionExpression='Records = :Records',
        ExpressionAttributeValues={
            ':Records': {'S': message_id}
        }
    )["Count"] != 0

# Processes the message body to upper case.
# @input string body
#
# @param body to be processed
# @return uppercase body
def process_message(body):
    return body.upper()

# Put the message to the DynamoDB Table.
# @input string batch_item_success
#
# @param batch_item_success of the message to put.
# @return Boolean
def push_to_dynamoDB(batch_item_success):

    for message_id in batch_item_success:
        response = dynamodb_client.put_item(
                        TableName = DDB_TABLE, 
                        Item={ 'Records': {'S':message_id}
                        }
                    )
    return True

# processor function iterating through messages in the event.
# @input dict Records
#
# @param Records to be processed
# @return Boolean
def iterate_records(Records):

    batch_item_failures = []
    batch_item_success = []

    for record in Records:

        message_id = record["messageId"]
        print("Message: " + message_id)
        if is_duplicate_message(message_id):   
            print("Message duplicate: " + message_id)
            continue

        try:
            process_message(record["body"])
            batch_item_success.append(message_id) 
        except:
            batch_item_failures.append({"itemIdentifier": message_id}) 

    push_to_dynamoDB(batch_item_success)
    return batch_item_failures

def lambda_handler(event, context):

    return {"batchItemFailures": iterate_records(event["Records"])}

Related information

Why did my Lambda function retry valid Amazon SQS messages and place them in my dead-letter queue?

Why doesn't my Amazon SQS queue invoke my Lambda function?

Why are my Amazon SQS messages unexpectedly sent to a DLQ?