Come posso impedire a un messaggio Amazon SQS di invocare la mia funzione Lambda più di una volta?

Breve descrizione

Lambda supporta la consegna at-least-once dei messaggi. In casi specifici, il meccanismo di ripetizione potrebbe inviare duplicati dello stesso messaggio. In seguito, Amazon SQS inviai messaggi alla tua coda DLQ, se ne hai configurata una.

Le invocazioni Lambda duplicate per lo stesso messaggio Amazon SQS possono verificarsi per uno dei seguenti motivi:

• La funzione restituisce un errore o si verifica il timeout.
• Il servizio Lambda ha elaborato correttamente il batch, ma non è riuscito a eliminare il messaggio dalla coda di Amazon SQS prima del termine del timeout di visibilità.
• Il servizio Lambda ha inviato l’evento alla funzione, ma non ha ricevuto una conferma da quest’ultima.
• A causa di un problema intermittente, Amazon SQS ha restituito lo stesso messaggio, che viene nuovamente interrogato dal servizio Lambda.
• La somma della finestra dei batch e della durata della funzione supera il timeout di visibilità della coda di Amazon SQS. Il timeout di visibilità SQS deve essere almeno sei volte il timeout totale della funzione e il timeout della finestra dei batch.

Per confermare di aver ricevuto lo stesso messaggio più volte, controlla l’ID del messaggio. Se ricevi più copie dello stesso messaggio, allora i messaggi hanno lo stesso ID.
Nota: se ricevi più messaggi con lo stesso contenuto ma con diversi ID dei messaggi, la coda ha ricevuto il messaggio più di una volta. Verifica che il produttore sia in esecuzione, invii i messaggi all’URL della coda corretto e disponga delle autorizzazioni richieste.

Risoluzione

Per risolvere questo problema, attiva la funzione Segnala errore di un elemento in batch nel trigger SQS della funzione Lambda. Quindi, crea un codice funzionale modulare che itera nel batch, elabora ed elimina i messaggi con esito positivo e duplicati. La funzione memorizza il message ID dei messaggi con esito positivo in una tabella di Amazon DynamoDB e quindi verifica che il messaggio sia stato elaborato in precedenza.

Importante: la risoluzione seguente rallenta i tempi di elaborazione inviando più richieste a DynamoDB per ogni messaggio in arrivo, pertanto i costi per le chiamate API saranno più elevati.

Creazione di una tabella DynamoDB

La tabella DynamoDB di seguito contiene gli ID dei messaggi in modo che una funzione Lambda possa confrontarli per la duplicazione.

Completa i seguenti passaggi:

1. Apri la console DynamoDB.
2. Scegli Crea tabella.
3. Nella pagina Crea tabella DynamoDB, inserisci i seguenti valori:
   Per Tabella, inserisci ProcessedRecords
   In Chiave di partizione, inserisci Records
   Imposta il tipo di dati su String
4. Scegli Crea tabella.

Crea una funzione Lambda

Importante: il codice della funzione Lambda deve essere idempotente. Per un esempio della logica idempotente delle funzioni, consulta Best practice per l’utilizzo delle funzioni AWS Lambda.

Crea una funzione Lambda che confronti i messaggi in arrivo con i messaggi con esito positivo in precedenza e poi conservati nella tabella DynamoDB. Se un messaggio ha avuto un esito positivo in precedenza, la funzione non consente di elaborare i duplicati. Se i nuovi messaggi univoci hanno esito positivo, vengono inviati alla tabella per un confronto futuro.

SQS effettua riprova a inviare i messaggi senza esito positivo finché Lambda non li elabora correttamente o finché il valore ReceiveCount di un messaggio non supera maxReceiveCount.

Nella tua funzione, aggiungi un ruolo di esecuzione per consentire le azioni dynamodb:Query e dynamodb:PutItem. Quindi, segnala gli errori degli elementi in batch nel trigger SQS per identificare e ignorare i messaggi duplicati nel batch.

L’esempio seguente di logica della funzione Lambda converte il corpo del messaggio in maiuscolo:

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

Informazioni correlate

Perché la mia funzione Lambda riprova a inviare i messaggi Amazon SQS validi e li inserisce nella mia coda DLQ?

Perché la mia coda Amazon SQS non invoca la mia funzione Lambda?

Perché i miei messaggi Amazon SQS vengono inviati inaspettatamente a un DLQ?