0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@sponge841841(ちっち)

Lambda × SQSで1件の失敗に全件巻き込まれない：ReportBatchItemFailuresをJavaで実装する

0
Posted at

Amazon SQSをAWS Lambdaのイベントソースにすると、Lambdaはメッセージを1件ずつではなくバッチで受け取ります。

ここで意外とハマりやすいのが、10件中9件の処理に成功していても、1件の失敗でLambda関数全体を例外終了させると、デフォルトではバッチ全体が再試行対象になることです。

その結果、すでに成功したメッセージまで再処理される可能性があります。

この記事では、AWS LambdaのReportBatchItemFailuresを使って、失敗したSQSメッセージだけを再試行対象にする方法をJavaで整理します。

まず結論

SQS → Lambdaのバッチ処理では、次の3点をセットで考えるのがおすすめです。

  1. イベントソースマッピングでReportBatchItemFailuresを有効にする
  2. Lambda側で各メッセージの例外を捕捉し、失敗したmessageIdだけ返す
  3. それでも重複処理は起こり得るので、処理自体を冪等にする

特に重要なのは、コードだけ変更しても不十分という点です。

Lambdaが部分的な失敗レスポンスを解釈するには、イベントソースマッピング側でもReportBatchItemFailuresを有効にする必要があります。

デフォルトでは何が起きるのか

たとえば、LambdaがSQSから次の5件をまとめて受け取ったとします。

id1 → 成功
id2 → 成功
id3 → 失敗
id4 → 成功
id5 → 成功

ここでid3の処理中に例外が外まで伝播し、Lambda関数自体が失敗すると、デフォルトではバッチ全体が失敗として扱われます。

つまり、id1、id2、id4、id5も再び処理される可能性があります。

外部API呼び出し、メール送信、決済、DB更新などを行っている場合、再処理による二重実行を考慮しなければなりません。

ReportBatchItemFailuresとは

ReportBatchItemFailuresは、バッチ内で失敗したメッセージだけをLambdaへ知らせるための仕組みです。

レスポンスは次のような形になります。

{
  "batchItemFailures": [
    {
      "itemIdentifier": "id3"
    }
  ]
}

この例ではid3だけを失敗として返しています。

AWS Lambdaは、成功したメッセージを再試行対象から外し、失敗したメッセージだけを再度処理できるようにします。

Javaで実装する

AWS公式ドキュメントでは、Java向けにSQSEventとSQSBatchResponseを使った実装例が紹介されています。

Java 17でも次のような形で実装できます。

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 SqsHandler implements RequestHandler<SQSEvent, SQSBatchResponse> {

    @Override
    public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
        List<SQSBatchResponse.BatchItemFailure> failures = new ArrayList<>();

        for (SQSEvent.SQSMessage message : event.getRecords()) {
            try {
                process(message);
            } catch (Exception e) {
                context.getLogger().log(
                    "Failed messageId=" + message.getMessageId()
                        + " error=" + e.getMessage()
                );

                failures.add(
                    new SQSBatchResponse.BatchItemFailure(
                        message.getMessageId()
                    )
                );
            }
        }

        return new SQSBatchResponse(failures);
    }

    private void process(SQSEvent.SQSMessage message) {
        if (message.getBody() == null || message.getBody().isBlank()) {
            throw new IllegalArgumentException("message body is empty");
        }

        // 業務処理
        System.out.println(message.getBody());
    }
}

ポイントは、メッセージ単位でtry-catchすることです。

やってはいけない例

@Override
public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
    for (SQSEvent.SQSMessage message : event.getRecords()) {
        process(message);
    }

    return new SQSBatchResponse(List.of());
}

process()が1件でも例外を投げ、その例外を捕捉しなければ、Lambda関数全体が例外終了します。

AWS公式ドキュメントでも、関数が例外をスローした場合はバッチ全体が失敗として扱われると説明されています。

そのため、部分バッチレスポンスを使うなら、失敗したメッセージを収集して正常にレスポンスを返す必要があります。

Lambda側の設定も必要

Javaコードを変更しただけではReportBatchItemFailuresは有効になりません。

イベントソースマッピングのFunctionResponseTypesにReportBatchItemFailuresを指定します。

AWS CLIなら次のように設定できます。

aws lambda update-event-source-mapping \
  --uuid "EVENT_SOURCE_MAPPING_UUID" \
  --function-response-types "ReportBatchItemFailures"

コマンドの意味

  • aws lambda update-event-source-mapping
    • Lambdaのイベントソースマッピング設定を変更するコマンド
  • --uuid
    • SQSとLambdaを紐付けているイベントソースマッピングの識別子
  • --function-response-types "ReportBatchItemFailures"
    • Lambdaから返された部分失敗レスポンスを有効にする設定

UUIDが分からない場合は、次のコマンドで確認できます。

aws lambda list-event-source-mappings \
  --function-name my-function

AWS SAMでは次のように指定できます。

Resources:
  WorkerFunction:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: java17
      Handler: example.SqsHandler::handleRequest
      Events:
        QueueEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt WorkQueue.Arn
            BatchSize: 10
            FunctionResponseTypes:
              - ReportBatchItemFailures

AWS SAMの公式仕様でも、SQSイベントのFunctionResponseTypesで指定できる値としてReportBatchItemFailuresが定義されています。

可視性タイムアウトにも注意する

SQSには**可視性タイムアウト（Visibility Timeout）**があります。

これは、あるコンシューマーがメッセージを処理している間、そのメッセージを他のコンシューマーから一時的に見えなくする時間です。

AWSは、SQSをLambdaのイベントソースにする場合、キューの可視性タイムアウトをLambda関数のタイムアウトの少なくとも6倍に設定することを推奨しています。

たとえばLambdaのタイムアウトが30秒なら、目安は次のようになります。

30秒 × 6 = 180秒

さらにバッチウィンドウを設定している場合、AWSは次の考え方を推奨しています。

可視性タイムアウト
= Lambdaタイムアウト × 6
+ MaximumBatchingWindowInSeconds

Lambdaが処理中なのに可視性タイムアウトが短すぎると、同じメッセージが再び見える状態になり、重複処理の原因になり得ます。

部分バッチレスポンスを使っても「exactly once」にはならない

ここは重要です。

ReportBatchItemFailuresを使えば、成功したメッセージの不要な再試行を減らせます。

ただし、SQSとLambdaの仕組み上、同じメッセージが複数回処理される可能性そのものをゼロにはできません。

そのため、処理は**冪等（idempotent）**にしておく必要があります。

冪等とは、同じ処理を複数回実行しても結果が変わらないように設計することです。

たとえば、注文IDをキーにDBへ処理済み記録を保存する方法があります。

1回目
order-123 → 未処理 → 実行 → 処理済みとして保存

2回目
order-123 → 処理済み → スキップ

メッセージIDだけではなく、業務上の一意キーを使う方が扱いやすいケースもあります。

DLQもセットで考える

何度再試行しても失敗するメッセージ、いわゆるポイズンメッセージが存在すると、同じ失敗が繰り返されます。

そこでSQSではDLQ（Dead-Letter Queue）を設定できます。

DLQは、一定回数以上処理に失敗したメッセージを隔離するためのキューです。

AWSはLambdaとSQSを組み合わせる際、ソースキューのRedrive PolicyのmaxReceiveCountを少なくとも5にすることを推奨しています。

通常キュー
   ↓
Lambda
   ↓ 失敗
再試行
   ↓
maxReceiveCount超過
   ↓
DLQ

部分バッチレスポンスとDLQは役割が違います。

  • ReportBatchItemFailures
    • 成功済みメッセージの不要な再試行を減らす
  • DLQ
    • 何度やっても失敗するメッセージを隔離する

両方を組み合わせるのが実運用では扱いやすいです。

FIFOキューでは処理方法が変わる

FIFOキューではメッセージの順序が重要です。

AWSは、FIFOキューで部分バッチレスポンスを使う場合、最初の失敗が発生した時点で後続メッセージの処理を止め、失敗したメッセージと未処理のメッセージをbatchItemFailuresへ含めることを推奨しています。

Standard Queueと同じ感覚で後続メッセージまで処理すると、順序保証を崩す可能性があります。

実務でのチェックリスト

SQS → Lambdaを実装するときは、最低限このあたりを確認しています。

  • ReportBatchItemFailuresを有効にしたか
  • メッセージ単位で例外を捕捉しているか
  • 失敗したmessageIdだけ返しているか
  • LambdaのタイムアウトとSQSの可視性タイムアウトを確認したか
  • DLQを設定したか
  • maxReceiveCountを決めたか
  • 同じメッセージが複数回来ても安全な処理になっているか
  • FIFOの場合は最初の失敗後に処理を止めているか

まとめ

LambdaとSQSの組み合わせでは、単純に「例外が起きたらthrowする」だけだと、成功済みのメッセージまで再試行されることがあります。

ReportBatchItemFailuresを使えば、失敗したメッセージだけをLambdaへ報告できます。

ただし、本番運用ではそれだけで終わりではありません。

部分バッチレスポンス
        +
冪等性
        +
Visibility Timeout
        +
DLQ

この4つをセットで設計すると、SQS → Lambdaの失敗処理をかなり整理しやすくなります。

参考資料

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?