Comment puis-je empêcher un message Amazon SQS d’invoquer ma fonction Lambda plusieurs fois ?

Brève description

Lambda prend en charge la livraison de messages au moins une fois. Dans certains cas, le mécanisme de nouvelle tentative peut envoyer des doublons du même message. Amazon SQS envoie ensuite les messages à votre file d’attente de lettres mortes, si vous en avez configuré une.

Des invocations Lambda en double pour le même message Amazon SQS peuvent se produire pour l’une des raisons suivantes :

• Votre fonction renvoie une erreur ou expire.
• Le service Lambda a traité le lot avec succès, mais n’a pas réussi à supprimer le message de la file d’attente Amazon SQS avant l’expiration du délai de visibilité.
• Le service Lambda a envoyé l’événement à la fonction, mais n’a pas reçu d’accusé de réception de la fonction.
• Suite à un problème intermittent, Amazon SQS a renvoyé le même message, qui a été à nouveau interrogé par le service Lambda.
• La somme de la fenêtre de traitement par lots et de la durée de la fonction est supérieure au délai de visibilité de votre file d’attente Amazon SQS. Le délai de visibilité SQS doit correspondre à au moins six fois la somme du délai d’expiration de la fonction et du délai d’expiration de la fenêtre de traitement par lots.

Pour confirmer que vous avez reçu le même message plusieurs fois, vérifiez l’ID du message. Si vous recevez plusieurs copies du même message, les messages ont le même ID.
Remarque : Si vous recevez plusieurs messages ayant le même contenu mais des ID de message différents, cela signifie que la file d’attente a reçu le message plusieurs fois. Vérifiez que le producteur s’exécute, qu’il envoie les messages à l’URL de file d’attente correcte et qu’il dispose des autorisations requises.

Résolution

Pour résoudre ce problème, activez l’option Signaler les échecs des éléments du lot dans le déclencheur SQS de votre fonction Lambda. Créez ensuite un code de fonction modulaire qui parcourt le lot, puis traite et supprime les messages réussis et les doublons. La fonction enregistre le messageID des messages réussis dans une table Amazon DynamoDB, puis vérifie qu’elle a déjà traité le message.

Important : La résolution suivante ralentit le temps de traitement, car elle envoie plusieurs demandes à DynamoDB pour chaque message entrant, et elle entraîne une augmentation des coûts liés aux appels d’API.

Créer une table DynamoDB

La table DynamoDB suivante contient les ID de vos messages afin qu’une fonction Lambda puisse les comparer pour détecter les doublons.

Procédez comme suit :

1. Ouvrez la console DynamoDB.
2. Sélectionnez Créer une table.
3. Sur la page Créer une table DynamoDB, saisissez les valeurs suivantes :
   Pour Table, saisissez ProcessedRecords
   Sous Clé de partition, saisissez Records
   Définissez le type de données sur String
4. Sélectionnez Créer une table.

Créer une fonction Lambda

Important : Le code de la fonction Lambda doit être idempotent. Pour obtenir un exemple de logique idempotente de fonction, consultez Bonnes pratiques d’utilisation des fonctions AWS Lambda.

Créez une fonction Lambda qui compare les messages entrants aux messages précédemment traités avec succès, puis conservés dans votre table DynamoDB. Si un message a déjà été traité avec succès, la fonction n’autorise pas le traitement des doublons. Si de nouveaux messages uniques sont traités avec succès, ils sont envoyés dans la table pour une comparaison ultérieure.

SQS effectue de nouvelles tentatives pour les messages en échec jusqu’à ce que Lambda les traite avec succès ou jusqu’à ce que le ReceiveCount d’un message dépasse le maxReceiveCount.

Dans votre fonction, ajoutez un rôle d’exécution pour autoriser les actions dynamodb:Query et dynamodb:PutItem. Signalez ensuite les échecs des éléments du lot dans le déclencheur SQS pour identifier et ignorer les messages en double dans le lot.

L’exemple de logique de fonction Lambda suivant convertit le corps du message en majuscules :

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"])}

Informations connexes

Pourquoi ma fonction Lambda réessaie-t-elle des messages Amazon SQS valides et les place-t-elle dans ma file d’attente de lettres mortes ?

Pourquoi ma file d’attente Amazon SQS n’invoque-t-elle pas ma fonction Lambda ?

Pourquoi mes messages Amazon SQS sont-ils envoyés de manière inattendue dans une DLQ ?