Amazon SQS 트리거가 있는 Lambda 함수의 배치 항목 실패 보고

Amazon SQS 트리거가 있는 Lambda 함수의 배치 항목 실패 보고

다음 코드 예제는 SQS 큐에서 이벤트를 받는 Lambda 함수에 대한 부분 배치 응답(partial batch response)을 구현하는 방법을 보여줘요. 이 함수는 응답에서 배치 항목 실패를 보고해 Lambda가 나중에 해당 메시지를 재시도하도록 신호를 보내요.

출처: AWS Lambda 개발자 안내서

본문

[ .NET ]

SDK for .NET
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
.NET으로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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 for Go V2
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
Go로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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 for Java 2.x
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
Java로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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 for JavaScript (v3)
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
JavaScript로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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}`);
}

TypeScript로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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 for PHP
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
PHP로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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 for Python (Boto3)
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
Python으로 Lambda에서 SQS 배치 항목 실패 보고하기.

# 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 for Ruby
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
Ruby로 Lambda에서 SQS 배치 항목 실패 보고하기.

# 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 for Rust
GitHub에서 더 많은 내용을 볼 수 있어요. Serverless examples 저장소에서 전체 예제를 찾고 설정·실행 방법을 배워 보세요.
Rust로 Lambda에서 SQS 배치 항목 실패 보고하기.

// 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
}

AWS SDK 개발자 안내서와 코드 예제 전체 목록은 Lambda를 AWS SDK와 함께 사용을 참고하세요. 이 주제에는 시작하기 정보와 이전 SDK 버전에 대한 세부 정보도 포함돼요.

더 알아보기 (Learn more)

• .NET·Go·Java·JavaScript·TypeScript·PHP·Python·Ruby·Rust로 SQS 트리거의 배치 항목 실패를 보고해 실패 메시지를 재시도하는 부분 배치 응답을 구현하는 방법을 익혀 보세요.