Stratégie d’interruption pour les invocations ayant échoué Mise en œuvre de réponses partielles de lot

Gestion des erreurs pour une source d’événements SQS dans Lambda

Pour gérer les erreurs liées à une source d’événement SQS, Lambda utilise automatiquement une stratégie de nouvelle tentative associée à une stratégie de backoff. Vous pouvez également personnaliser le comportement de gestion des erreurs en configurant votre mappage des sources d’événements SQS de manière à renvoyer des réponses par lots partielles.

Stratégie d’interruption pour les invocations ayant échoué

Lorsqu’une invocation échoue, Lambda tente de réessayer l’invocation tout en mettant en œuvre une stratégie d’interruption. La stratégie d’interruption diffère légèrement selon que Lambda a rencontré l’échec à cause d’une erreur dans votre code de fonction, ou à cause de la limitation.

Si votre code de fonction est à l'origine de l'erreur, Lambda arrête le traitement et réessaie l'invocation. Dans le même temps, Lambda réduit progressivement les tentatives en diminuant la quantité de simultanéité allouée à votre mappage des sources d’événements Amazon SQS. Une fois le délai de visibilité de votre file d'attente expiré, le message réapparaît dans la file d'attente.
Si l’invocation échoue en raison d’une limitation, Lambda réduit progressivement les tentatives en diminuant la quantité de simultanéité allouée à votre mappage des sources d’événements Amazon SQS. Lambda continue à réessayer le message jusqu’à ce que l’horodatage du message dépasse le délai de visibilité de votre file d’attente, auquel cas Lambda abandonne le message.

Mise en œuvre de réponses partielles de lot

Lorsque votre fonction Lambda rencontre une erreur lors du traitement d’un lot, tous les messages de ce lot redeviennent par défaut visibles dans la file d’attente, y compris les messages traités avec succès par Lambda. Par conséquent, votre fonction peut finir par traiter le même message plusieurs fois.

Pour éviter de retraiter les messages traités avec succès d’un lot en échec, vous pouvez configurer le mappage des sources d’événements pour que seuls les messages ayant échoué soient à nouveau visibles. C’est ce que l’on appelle une réponse partielle de lot. Pour activer les réponses partielles par lots, spécifiez l'FunctionResponseTypesaction ReportBatchItemFailures à effectuer lors de la configuration de votre mappage de sources d'événements. Avec ce paramètre, votre fonction peut renvoyer un succès partiel, ce qui peut contribuer à réduire le nombre de nouvelles tentatives inutiles sur les enregistrements.

Note

L'utilitaire Batch Processor de Powertools pour AWS Lambda gère automatiquement toute la logique de réponse partielle par lots. Cet utilitaire simplifie la mise en œuvre de modèles de traitement par lots et réduit le code personnalisé nécessaire pour gérer correctement les échecs d’éléments de lots. Il est disponible pour Python, Java, TypeScript et .NET.

Lorsque ReportBatchItemFailures est activé, Lambda ne réduit pas la taille de l’interrogation des messages lorsque les invocations de fonctions échouent. Si vous vous attendez à ce que certains messages échouent et que vous ne voulez pas que ces échecs aient un impact sur le taux de traitement des messages, utilisez ReportBatchItemFailures.

Note

Gardez les points suivants à l’esprit lorsque vous utilisez des réponses partielles de lot :

Si la fonction génère une exception, l’ensemble du lot est considéré comme un échec complet.
Si vous utilisez cette fonctionnalité avec une file d’attente FIFO, votre fonction doit arrêter le traitement des messages après le premier échec et renvoyer tous les messages échoués et non traités dans batchItemFailures. Cela permet de préserver l’ordre des messages dans votre file d’attente.

Pour activer les rapports partiels de lot

Consultez les bonnes pratiques pour la mise en œuvre des réponses partielles de lot.
Exécutez la commande suivante pour activer ReportBatchItemFailures pour votre fonction. Pour récupérer l'UUID de votre mappage de sources d'événements, exécutez la commande https://docs.aws.amazon.com/cli/latest/reference/lambda/list-event-source-mappings.html AWS CLI list-event-source-mappings.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--function-response-types "ReportBatchItemFailures"
```

Mettez à jour le code de votre fonction afin de capturer toutes les exceptions et de renvoyer les messages d’échec dans une réponse JSON batchItemFailures. La réponse batchItemFailures doit inclure une liste d’identifiants de message, sous forme de valeurs JSON itemIdentifier.

Supposons par exemple que vous avez un lot de cinq messages avec des identifiants de message id1, id2, id3, id4 et id5. Votre fonction traite avec succès id1, id3 et id5. Pour rendre les messages id2 et id4 visibles de nouveau dans la file d’attente, votre fonction doit renvoyer la réponse suivante :


{ 
  "batchItemFailures": [ 
        {
            "itemIdentifier": "id2"
        },
        {
            "itemIdentifier": "id4"
        }
    ]
}

Voici quelques exemples de codes de fonctions qui renvoient la liste des identifiants des messages ayant échoué dans le lot :

.NET

SDK pour .NET

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de .NET.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;

// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace sqsSample;

public class Function
{
    public async Task<SQSBatchResponse> FunctionHandler(SQSEvent evnt, ILambdaContext context)
    {
        List<SQSBatchResponse.BatchItemFailure> batchItemFailures = new List<SQSBatchResponse.BatchItemFailure>();
        foreach(var message in evnt.Records)
        {
            try
            {
                //process your message
                await ProcessMessageAsync(message, context);
            }
            catch (System.Exception)
            {
                //Add failed message identifier to the batchItemFailures list
                batchItemFailures.Add(new SQSBatchResponse.BatchItemFailure{ItemIdentifier=message.MessageId}); 
            }
        }
        return new SQSBatchResponse(batchItemFailures);
    }

    private async Task ProcessMessageAsync(SQSEvent.SQSMessage message, ILambdaContext context)
    {
        if (String.IsNullOrEmpty(message.Body))
        {
            throw new Exception("No Body in SQS Message.");
        }
        context.Logger.LogInformation($"Processed message {message.Body}");
        // TODO: Do interesting work based on the new message
        await Task.CompletedTask;
    }
}

Go

Kit SDK pour Go V2

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de Go.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main

import (
	"context"
	"fmt"
	"github.com/aws/aws-lambda-go/events"
	"github.com/aws/aws-lambda-go/lambda"
)

func handler(ctx context.Context, sqsEvent events.SQSEvent) (map[string]interface{}, error) {
	batchItemFailures := []map[string]interface{}{}

	for _, message := range sqsEvent.Records {
		if len(message.Body) > 0 {
			// Your message processing condition here
			fmt.Printf("Successfully processed message: %s\n", message.Body)
		} else {
			// Message processing failed
			fmt.Printf("Failed to process message %s\n", message.MessageId)
			batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": message.MessageId})
		}
	}

	sqsBatchResponse := map[string]interface{}{
		"batchItemFailures": batchItemFailures,
	}
	return sqsBatchResponse, nil
}

func main() {
	lambda.Start(handler)
}

Java

SDK pour Java 2.x

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de Java.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
 
import java.util.ArrayList;
import java.util.List;
 
public class ProcessSQSMessageBatch implements RequestHandler<SQSEvent, SQSBatchResponse> {
    @Override
    public SQSBatchResponse handleRequest(SQSEvent sqsEvent, Context context) {
         List<SQSBatchResponse.BatchItemFailure> batchItemFailures = new ArrayList<SQSBatchResponse.BatchItemFailure>();

         for (SQSEvent.SQSMessage message : sqsEvent.getRecords()) {
             try {
                 //process your message
             } catch (Exception e) {
                 //Add failed message identifier to the batchItemFailures list
                 batchItemFailures.add(new SQSBatchResponse.BatchItemFailure(message.getMessageId()));
             }
         }
         return new SQSBatchResponse(batchItemFailures);
     }
}

JavaScript

SDK pour JavaScript (v3)

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signaler les défaillances d'articles par lots SQS avec Lambda à l'aide de. JavaScript


// Node.js 20.x Lambda runtime, AWS SDK for Javascript V3
export const handler = async (event, context) => {
    const batchItemFailures = [];
    for (const record of event.Records) {
        try {
            await processMessageAsync(record, context);
        } catch (error) {
            batchItemFailures.push({ itemIdentifier: record.messageId });
        }
    }
    return { batchItemFailures };
};

async function processMessageAsync(record, context) {
    if (record.body && record.body.includes("error")) {
        throw new Error("There is an error in the SQS Message.");
    }
    console.log(`Processed message: ${record.body}`);
}

Signaler les défaillances d'articles par lots SQS avec Lambda à l'aide de. TypeScript


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SQSEvent, SQSBatchResponse, Context, SQSBatchItemFailure, SQSRecord } from 'aws-lambda';

export const handler = async (event: SQSEvent, context: Context): Promise<SQSBatchResponse> => {
    const batchItemFailures: SQSBatchItemFailure[] = [];

    for (const record of event.Records) {
        try {
            await processMessageAsync(record);
        } catch (error) {
            batchItemFailures.push({ itemIdentifier: record.messageId });
        }
    }

    return {batchItemFailures: batchItemFailures};
};

async function processMessageAsync(record: SQSRecord): Promise<void> {
    if (record.body && record.body.includes("error")) {
        throw new Error('There is an error in the SQS Message.');
    }
    console.log(`Processed message ${record.body}`);
}

PHP

Kit SDK pour PHP

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de PHP.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
<?php

use Bref\Context\Context;
use Bref\Event\Sqs\SqsEvent;
use Bref\Event\Sqs\SqsHandler;
use Bref\Logger\StderrLogger;

require __DIR__ . '/vendor/autoload.php';

class Handler extends SqsHandler
{
    private StderrLogger $logger;
    public function __construct(StderrLogger $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @throws JsonException
     * @throws \Bref\Event\InvalidLambdaEvent
     */
    public function handleSqs(SqsEvent $event, Context $context): void
    {
        $this->logger->info("Processing SQS records");
        $records = $event->getRecords();

        foreach ($records as $record) {
            try {
                // Assuming the SQS message is in JSON format
                $message = json_decode($record->getBody(), true);
                $this->logger->info(json_encode($message));
                // TODO: Implement your custom processing logic here
            } catch (Exception $e) {
                $this->logger->error($e->getMessage());
                // failed processing the record
                $this->markAsFailed($record);
            }
        }
        $totalRecords = count($records);
        $this->logger->info("Successfully processed $totalRecords SQS records");
    }
}

$logger = new StderrLogger();
return new Handler($logger);

Python

Kit SDK for Python (Boto3)

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de Python.


# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

def lambda_handler(event, context):
    if event:
        batch_item_failures = []
        sqs_batch_response = {}
     
        for record in event["Records"]:
            try:
                print(f"Processed message: {record['body']}")
            except Exception as e:
                batch_item_failures.append({"itemIdentifier": record['messageId']})
        
        sqs_batch_response["batchItemFailures"] = batch_item_failures
        return sqs_batch_response

Ruby

Kit SDK pour Ruby

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de Ruby.


# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'json'

def lambda_handler(event:, context:)
  if event
    batch_item_failures = []
    sqs_batch_response = {}

    event["Records"].each do |record|
      begin
        # process message
      rescue StandardError => e
        batch_item_failures << {"itemIdentifier" => record['messageId']}
      end
    end

    sqs_batch_response["batchItemFailures"] = batch_item_failures
    return sqs_batch_response
  end
end

Rust

SDK pour Rust

Note

Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’exemples sans serveur.

Signalement des échecs d’articles par lots SQS avec Lambda à l’aide de Rust.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
    event::sqs::{SqsBatchResponse, SqsEvent},
    sqs::{BatchItemFailure, SqsMessage},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};

async fn process_record(_: &SqsMessage) -> Result<(), Error> {
    Err(Error::from("Error processing message"))
}

async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<SqsBatchResponse, Error> {
    let mut batch_item_failures = Vec::new();
    for record in event.payload.records {
        match process_record(&record).await {
            Ok(_) => (),
            Err(_) => batch_item_failures.push(BatchItemFailure {
                item_identifier: record.message_id.unwrap(),
            }),
        }
    }

    Ok(SqsBatchResponse {
        batch_item_failures,
    })
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    run(service_fn(function_handler)).await
}

Si les événements ayant échoué ne sont pas renvoyés dans la file d'attente, voir Comment résoudre les problèmes liés à la fonction Lambda SQS ? ReportBatchItemFailures dans le centre de AWS connaissances.

Conditions de réussite et d’échec

Lambda traite un lot comme un succès complet si votre fonction renvoie l’un des éléments suivants :

Une liste batchItemFailures vide
Une liste batchItemFailures nulle
Une EventResponse vide
Une EventResponse nulle

Lambda traite un lot comme un échec complet si votre fonction renvoie l’un des éléments suivants :

Une réponse JSON non valide
Une chaîne itemIdentifier vide
Un itemIdentifier nul
Un itemIdentifier avec un nom de clé incorrect
Une valeur itemIdentifier avec un ID de message inexistant

CloudWatch métriques

Pour déterminer si votre fonction signale correctement les défaillances d'articles par lots, vous pouvez surveiller les métriques NumberOfMessagesDeleted et ApproximateAgeOfOldestMessage Amazon SQS sur Amazon CloudWatch.

NumberOfMessagesDeleted suit le nombre de messages supprimés de votre file d’attente. Si cela tombe à 0, cela indique que la réponse de votre fonction ne renvoie pas correctement les messages d’échec.
ApproximateAgeOfOldestMessage suit combien de temps le message le plus ancien est resté dans votre file d’attente. Une forte augmentation de cette métrique peut indiquer que votre fonction ne renvoie pas correctement les messages d’échec.

Utilisation de Powertools pour AWS Lambda processeur par lots

L'utilitaire de traitement par lots de Powertools gère AWS Lambda automatiquement la logique de réponse partielle par lots, réduisant ainsi la complexité de la mise en œuvre des rapports de défaillance par lots. Voici des exemples d’utilisation de l’utilitaire de traitement par lots :

Python

Note

Pour des exemples complets et des instructions de configuration, consultez la documentation de l’utilitaire de traitement par lots.

Traitement des messages Amazon SQS à l'aide d'un processeur AWS Lambda par lots.


import json
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 import SQSEvent
from aws_lambda_powertools.utilities.typing import LambdaContext

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

def record_handler(record):
    logger.info(record)
    # Your business logic here
    # Raise an exception to mark this record as failed
    
def lambda_handler(event, context: LambdaContext):
    return process_partial_response(
        event=event, 
        record_handler=record_handler, 
        processor=processor,
        context=context
    )

TypeScript

Note

Pour des exemples complets et des instructions de configuration, consultez la documentation de l’utilitaire de traitement par lots.

Traitement des messages Amazon SQS à l'aide d'un processeur AWS Lambda par lots.


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

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

const recordHandler = async (record: any): Promise<void> => {
    logger.info('Processing record', { record });
    // Your business logic here
    // Throw an error to mark this record as failed
};

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

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Comportement de mise à l’échelle.

Parameters