Backoff-Strategie für fehlgeschlagene Aufrufe Implementierung von partiellen Batch-Antworten

Behandlung von Fehlern für eine SQS-Ereignisquelle in Lambda

Um Fehler im Zusammenhang mit einer SQS-Ereignisquelle zu behandeln, verwendet Lambda automatisch eine Wiederholungsstrategie mit einer Backoff-Strategie. Sie können das Verhalten bei der Fehlerbehandlung auch anpassen, indem Sie Ihre SQS-Zuordnung von Ereignisquellen so konfigurieren, dass sie partielle Batch-Antworten zurückgibt.

Backoff-Strategie für fehlgeschlagene Aufrufe

Wenn ein Aufruf fehlschlägt, versucht Lambda, den Aufruf zu wiederholen, während eine Backoff-Strategie implementiert wird. Die Backoff-Strategie unterscheidet sich geringfügig, je nachdem, ob Lambda den Fehler aufgrund eines Fehlers in Ihrem Funktionscode oder aufgrund einer Drosselung festgestellt hat.

Wenn Ihr Funktionscode den Fehler verursacht hat, stoppt Lambda die Verarbeitung und versucht erneut, den Aufruf durchzuführen. In der Zwischenzeit zieht sich Lambda allmählich zurück und reduziert die Menge der Gleichzeitigkeit, die Ihrer Amazon SQS-Zuordnung von Ereignisquellen zugewiesen wird. Nachdem das Sichtbarkeits-Timeout Ihrer Warteschlange abgelaufen ist, erscheint die Nachricht erneut in der Warteschlange.
Wenn der Aufruf aufgrund von Drosselung fehlschlägt, setzt Lambda schrittweise Wiederholungsversuche zurück, indem es die Menge der Gleichzeitigkeit reduziert, die Ihrer Zuordnung von Ereignisquellen in Amazon SQS zugewiesen ist. Lambda fährt fort, die Nachricht erneut zu versuchen, bis der Zeitstempel der Nachricht das Sichtbarkeits-Timeout Ihrer Warteschlange überschreitet, an welchem Punkt Lambda die Nachricht verwirft.

Implementierung von partiellen Batch-Antworten

Wenn die Lamda-Funktion während der Verarbeitung eines Batches auf einen Fehler stößt, werden standardmäßig alle Nachrichten in diesem Batch wieder in der Warteschlange sichtbar, einschließlich Nachrichten, die Lambda erfolgreich verarbeitet hat. Infolgedessen kann es passieren, dass die Funktion dieselbe Nachricht mehrmals verarbeitet.

Damit nicht erfolgreich bearbeiten Nachrichten in einem fehlgeschlagenen Batch erneut verarbeitet werden, können Sie die Zuordnung von Ereignisquellen so konfigurieren, dass nur die fehlgeschlagenen Nachrichten wieder sichtbar werden. Dies wird als partielle Batch-Antwort bezeichnet. Um partielle Batch-Antworten zu aktivieren, geben Sie ReportBatchItemFailures bei der Konfiguration der Ereignisquellenzuordnung die FunctionResponseTypes Aktion an. Mit dieser Einstellung kann Ihre Funktion einen teilweisen Erfolg zurückgeben, was dazu beitragen kann, die Anzahl unnötiger Wiederholungen bei Datensätzen zu reduzieren.

Anmerkung

Das Batch Processor-Hilfsprogramm von Powertools for AWS Lambda verarbeitet die gesamte partielle Batch-Antwortlogik automatisch. Dieses Tool vereinfacht die Implementierung von Batch-Verarbeitungsmustern und reduziert den Bedarf an benutzerdefiniertem Code, der für die korrekte Behandlung von Fehlern bei Batch-Elementen erforderlich ist. Es ist für Python, Java, Typescript und .NET verfügbar.

Wenn ReportBatchItemFailures aktiviert ist, skaliert Lambda die Nachrichtenabfrage nicht herunter, wenn Funktionsaufrufe fehlschlagen. Wenn Sie erwarten, dass einige Nachrichten ausfallen – und Sie nicht möchten, dass sich diese Fehler auf die Nachrichtenverarbeitungsrate auswirken – verwenden Sie ReportBatchItemFailures.

Anmerkung

Berücksichtigen Sie bei Verwendung von partiellen Batch-Antworten Folgendes:

Wenn die Funktion eine Ausnahme ausgibt, gilt der gesamte Batch als fehlgeschlagen.
Wenn Sie dieses Feature mit einer FIFO-Warteschlange verwenden, sollte Ihre Funktion die Verarbeitung von Nachrichten nach dem ersten Fehler beenden und alle fehlgeschlagenen und nicht verarbeiteten Nachrichten in batchItemFailures zurückgeben. Das hilft, die Reihenfolge der Nachrichten in der Warteschlange beizubehalten.

So aktivieren Sie partielle Batchberichterstattung

Sehen Sie sich die bewährten Methoden für die Implementierung von partiellen Batch-Antworten an.
Führen Sie den folgenden Befehl aus, um ReportBatchItemFailures für Ihre Funktion zu aktivieren. Führen Sie den Befehl list-event-source-mappings aus, um die UUID Ihres Eventquellen-Mappings abzurufen. https://docs.aws.amazon.com/cli/latest/reference/lambda/list-event-source-mappings.html AWS CLI
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--function-response-types "ReportBatchItemFailures"
```

Aktualisieren Sie Ihren Funktionscode, um alle Ausnahmen abzufangen und fehlgeschlagene Nachrichten in einer batchItemFailures-JSON-Antwort zurückzugeben. Die batchItemFailures-Antwort muss eine Liste von Nachrichten-IDs als itemIdentifier-JSON-Werte enthalten.

Angenommen, Sie haben einen Batch von fünf Nachrichten mit den Nachrichten-IDs id1, id2, id3, id4 und id5. Die Funktion verarbeitet erfolgreich id1, id3 und id5. Damit die Nachrichten id2 und id4 in der Warteschlange wieder sichtbar werden, sollte Ihre Funktion folgende Antwort zurückgeben:


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

Hier sind einige Beispiele für Funktionscodes, die die Liste der fehlgeschlagenen Nachrichten-IDs im Batch zurückgeben:

.NET

SDK für .NET

Anmerkung

GitHubEs läuft noch mehr. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von SQS-Batchelementfehlern mit Lambda unter Verwendung von .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

SDK für Go V2

Anmerkung

Es läuft noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von SQS-Batchelementfehlern mit Lambda unter Verwendung von 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 für Java 2.x

Anmerkung

Es läuft noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von Fehlern bei SQS-Batchelementen mit Lambda unter Verwendung von 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 für JavaScript (v3)

Anmerkung

Es gibt noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Meldung von SQS-Batch-Elementfehlern mithilfe von Lambda. 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}`);
}

Meldung von Fehlern bei SQS-Batch-Elementen unter Verwendung von Lambda. 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

SDK für PHP

Anmerkung

Es läuft noch mehr. GitHub Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von SQS-Batchelementfehlern mit Lambda unter Verwendung von 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

SDK für Python (Boto3)

Anmerkung

Es läuft noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von Fehlern bei SQS-Batchelementen mit Lambda unter Verwendung von 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

SDK für Ruby

Anmerkung

Es läuft noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von Fehlern bei SQS-Batchelementen mit Lambda unter Verwendung von 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 für Rust

Anmerkung

Es läuft noch mehr GitHub. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen.

Melden von Fehlern bei SQS-Batchelementen mit Lambda unter Verwendung von 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
}

Wenn die fehlgeschlagenen Ereignisse nicht in die Warteschlange zurückkehren, finden Sie weitere Informationen unter Wie behebe ich die Lambda-Funktion ReportBatchItemFailures SQS? im AWS Knowledge Center.

Erfolgs- und Misserfolgsbedingungen

Lambda behandelt einen Batch als komplett erfolgreich, wenn die Funktion eines der folgenden Elemente zurückgibt:

Eine leere batchItemFailures-Liste
Eine ungültige batchItemFailures-Liste
Ein leeres EventResponse
Ein ungültiges EventResponse

Lambda behandelt einen Batch als komplett fehlgeschlagen, wenn die Funktion eines der folgenden Elemente zurückgibt:

Eine ungültige JSON-Antwort
Eine leere Zeichenfolge itemIdentifier
Ein ungültiges itemIdentifier
Ein itemIdentifier mit einem falschen Schlüsselnamen
Einen itemIdentifier-Wert mit einer Nachrichten-ID, die nicht existiert

CloudWatch Metriken

Um festzustellen, ob Ihre Funktion Fehler bei Chargenartikeln korrekt meldet, können Sie die NumberOfMessagesDeleted und ApproximateAgeOfOldestMessage Amazon SQS-Metriken in Amazon CloudWatch überwachen.

NumberOfMessagesDeleted verfolgt die Anzahl der Nachrichten, die aus der Warteschlange entfernt wurden. Wenn der Wert auf 0 sinkt, ist dies ein Zeichen dafür, dass die Funktionsantwort fehlgeschlagene Nachrichten nicht korrekt zurückgibt.
ApproximateAgeOfOldestMessage verfolgt, wie lange die älteste Nachricht in der Warteschlange geblieben ist. Ein starker Anstieg dieser Metrik kann darauf hinweisen, dass die Funktion fehlgeschlagene Nachrichten nicht korrekt zurückgibt.

Verwenden Sie Powertools für AWS Lambda Batch-Prozessor

Das Batch-Verarbeitungsprogramm von Powertools for verarbeitet AWS Lambda automatisch die partielle Batch-Antwortlogik und reduziert so die Komplexität der Implementierung von Batch-Fehlerberichten. Hier sind Beispiele für die Verwendung des Batch-Prozessors:

Python

Anmerkung

Umfassende Beispiele und Anweisungen zur Einrichtung finden Sie in der Dokumentation zum Batch-Prozessor.

Verarbeitung von Amazon SQS-Nachrichten mit einem AWS Lambda Batch-Prozessor.


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

Anmerkung

Umfassende Beispiele und Anweisungen zur Einrichtung finden Sie in der Dokumentation zum Batch-Prozessor.

Verarbeitung von Amazon SQS-Nachrichten mit AWS Lambda Batch-Prozessor.


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,
    });
};

Warnung JavaScript ist in Ihrem Browser nicht verfügbar oder deaktiviert.

Zur Nutzung der AWS-Dokumentation muss JavaScript aktiviert sein. Weitere Informationen finden auf den Hilfe-Seiten Ihres Browsers.

Dokumentkonventionen

Skalierungsverhalten

Parameters