Claude CodeでKinesisとLambdaの毒レコード詰まりを再現する

Kinesis Data StreamsのレコードをLambdaで処理する構成は、既定のままだと「処理できないレコードが1件あるだけで、そのシャードが止まる」作りです。AWSのドキュメントによれば、既定の設定では壊れたレコードがシャードの処理を最長1週間ふさぎます。

この挙動は、本番で起きてから読むより、使い捨ての環境で一度起こしておくほうが対処を決めやすくなります。ここでは、再現用のストリームとLambdaをTerraformで組む作業をClaude Codeに任せ、設定を段階的に変えながら詰まり方がどう変わるかを見ていきます。

既定の設定だと、なぜシャードが止まるのか

Lambdaは、Kinesisのレコードをバッチでまとめて関数に渡します。1つのバッチに入るのは、1つのシャードのレコードだけです。シャード内の順序は守られるので、関数がエラーを返すと、Lambdaはそのシャードの先へ進まず、同じバッチをやり直します。

やり直しを止める条件は、設定で決まります。ドキュメントにある条件は次の3つです。

• レコードの有効期限が切れる
• MaximumRecordAgeInSecondsで決めた最大経過時間を超える
• MaximumRetryAttemptsで決めた再試行の上限に達する

どちらの設定も既定値は-1で、無制限の意味です。つまり既定では、Lambda側に諦める条件がありません。ドキュメントも、既定では壊れたレコードがシャードの処理を最長1週間ふさぐと説明しています。

止まるのは、毒レコードのあるシャードです。シャードが多いほど同時に処理されるバッチが増え、エラーの影響は小さくなります。逆にシャード数が少ない構成ほど、止まる範囲は全体に近づきます。

呼び出す前の失敗と、呼び出し中の失敗は別物

ドキュメントは、エラーを2つに分けています。関数を呼び出す前にスロットリングなどで失敗した場合も、呼び出し後に関数がエラーを返した場合も、Lambdaは最大経過時間まで再試行します。

後者には追加の手段があります。BisectBatchOnFunctionErrorで、失敗したバッチを2つに割って再実行できます。割る操作は、再試行の回数に数えられません。

4つの設定が、それぞれ何を変えるか

再現に使う設定を、役割で並べます。既定値はKinesisのパラメーター一覧に載っているものです。

FunctionResponseTypesにReportBatchItemFailuresを入れる方法もあります。関数が「どのレコードで失敗したか」を返せるので、成功した分までやり直さずに済みます。

毒レコードを再現する環境を組む

再現の方針は単純です。シャード1本のストリームを作り、JSONのpoisonが真のレコードで例外を出す関数をつなぎます。シャードが1本なら、毒レコードの後ろに積んだ正常なレコードが処理されない様子が、そのまま見えます。

関数のコードは、Pythonで次のように書けます。段階を進めるまでは、例外をそのまま上げます。

import base64, json
 
def handler(event, context):
    failures = []
    for r in event["Records"]:
        body = json.loads(base64.b64decode(r["kinesis"]["data"]))
        if body.get("poison"):
            # 段階3では例外を上げず、失敗位置を返す
            raise ValueError("poison record")
    return {"batchItemFailures": failures}

Terraformでは、イベントソースマッピングが肝になります。引数名は、APIのパラメーターをsnake_caseにした形で指定します。段階ごとに値を変える前提で、最初は何も指定しません。

resource "aws_kinesis_stream" "poison_lab" {
  name        = "poison-lab"
  shard_count = 1
}
 
resource "aws_lambda_event_source_mapping" "poison_lab" {
  event_source_arn  = aws_kinesis_stream.poison_lab.arn
  function_name     = aws_lambda_function.consumer.arn
  starting_position = "TRIM_HORIZON"
  batch_size        = 10
  # 段階1: 上の設定のみ(再試行も最大経過時間も無制限)
}

batch_sizeを小さく置くのは、バッチが割れる様子を見やすくするためです。KinesisのBatchSizeは既定で100、最大で10,000です。

段階を3つに分けて、設定を足していく

詰まり方の違いは、設定を足すたびに1つずつ確かめるほうが読み取りやすくなります。

段階2のTerraformは、次の差分になります。再試行は2回、最大経過時間は1時間という、ドキュメントのCLI例と同じ値を使います。

resource "aws_lambda_event_source_mapping" "poison_lab" {
  # 段階1の設定はそのまま
  maximum_retry_attempts         = 2
  maximum_record_age_in_seconds  = 3600
 
  destination_config {
    on_failure {
      destination_arn = aws_sqs_queue.poison_dlq.arn
    }
  }
}

送信先のSQSキューを使うには、関数の実行ロールにsqs:SendMessageが必要です。SNSならsns:Publish、S3ならs3:PutObjectとs3:ListBucketを許可します。

SQSの失敗先に入るのは、レコード本体ではない

ここは読み違えやすい点です。SQSとSNSに届くのは、失敗の呼び出しに関するメタデータだけで、レコード本体は含まれません。含まれるのは、streamArn、shardId、startSequenceNumber、endSequenceNumberなどです。

本体が欲しいときは、この4つの値を使って、有効期限が切れる前にストリームから読み直します。S3を失敗先にすれば、メタデータに加えてpayloadとして元の呼び出しの内容も保存されます。

毒レコードの中身を後から調べたい場合は、S3のほうが手間が少なくなります。ただしバケットを消した後も設定が残ると、同名のバケットを別アカウントに作られて、失敗データが流れる危険があります。実行ロールのs3:PutObjectは、自分のアカウントのバケットに絞ります。

段階3:失敗位置を返し、バッチを割る

ReportBatchItemFailuresを使うには、イベントソースマッピングのfunction_response_typesにReportBatchItemFailuresを入れます。関数側は、失敗したレコードのシーケンス番号をitemIdentifierで返します。

ストリームは順序が大事なので、ドキュメントの.NETの例は、最初に失敗したレコードを見つけた時点で、その1件だけを返して処理を打ち切っています。Lambdaはその位置から先をやり直します。先ほどの関数は、次のように直せます。

import base64, json
 
def handler(event, context):
    for r in event["Records"]:
        body = json.loads(base64.b64decode(r["kinesis"]["data"]))
        if body.get("poison"):
            seq = r["kinesis"]["sequenceNumber"]
            return {"batchItemFailures": [{"itemIdentifier": seq}]}
    return {"batchItemFailures": []}

Terraform側は、次の2行を足します。

  bisect_batch_on_function_error = true
  function_response_types        = ["ReportBatchItemFailures"]

両方を有効にしたときの動きは、ドキュメントに書かれています。部分的な成功の応答を受けたLambdaは、返されたシーケンス番号の位置でバッチを割り、残りのレコードだけを再試行します。

BisectBatchOnFunctionErrorだけを有効にした場合も、関数が失敗すればバッチは割られます。この割れ方は、ReportBatchItemFailuresの設定に関係しません。

返答の形にも条件があります。itemIdentifierが空文字列、null、キー名の誤りのいずれかだと、Lambdaはバッチ全体の失敗として扱います。逆に、batchItemFailuresが空のリストなら、全体の成功です。

割っても、毒レコードの1件は残る

バッチを割る目的は、毒レコードを正常なレコードから切り離すことです。毒レコード自体は、分割の後も失敗し続けます。最後に毒レコード1件だけのバッチになり、そこからMaximumRetryAttemptsまでの再試行を重ねて、上限に達した時点で失敗先へ送られます。

だから段階3は、段階2を置き換えるのではなく、重ねて使う設定です。再試行の上限を決めずに分割だけを有効にしても、詰まる対象が1件に減るだけで、その1件が最長1週間居座る点は変わりません。

Claude Codeに任せる範囲を、フックで先に決める

この再現環境は、実際にAWSのリソースを作ります。Claude Codeにそのままterraform applyを実行させると、手元の認証情報が指す先に作られます。サンドボックスではないアカウントで走らせないため、実行前に止めるフックを置きます。

フックは、.claude/settings.jsonに書きます。matcherでBashツールを選び、ifでterraform applyとaws kinesis put-recordのコマンドだけに絞ります。

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "if": "Bash(terraform apply *)",
            "command": "${CLAUDE_PROJECT_DIR}/.claude/hooks/sandbox-only.sh"
          },
          {
            "type": "command",
            "if": "Bash(aws kinesis put-record *)",
            "command": "${CLAUDE_PROJECT_DIR}/.claude/hooks/sandbox-only.sh"
          }
        ]
      }
    ]
  }
}

スクリプトは、環境変数AWS_PROFILEが検証用のプロファイルでなければ、拒否の判断をJSONで返します。

#!/bin/bash
if [ "$AWS_PROFILE" != "sandbox" ]; then
  cat <<'JSON'
{
  "hookSpecificOutput": {
    "hookEventName": "PreToolUse",
    "permissionDecision": "deny",
    "permissionDecisionReason": "sandbox 以外のプロファイルでは実行できません"
  }
}
JSON
fi
exit 0

条件を満たすときは何も出力せずにexit 0します。この場合、フックは判断を出さないだけで承認もしないので、通常の権限確認に進みます。フックの仕組みの全体は、Claude Code Hooks完全ガイドにあります。

権限のdenyではなく、フックにする理由

権限ルールでBash(aws *)を拒否にすると、より狭い許可ルールがあっても拒否が勝ちます。aws kinesis put-recordだけを許可して残りを拒否する、という書き分けは通りません。

また、AWS_PROFILEのような実行時の条件で通す・止めるを切り替えるには、スクリプトで判断できるフックが向いています。フックの失敗でも止めたい場合は、onFailureで操作を止める設定で補えます。権限ルールの書き方は、パラメーターで書く権限ルールも参考になります。

再現の依頼文と、確かめる順番

Claude Codeへの依頼は、段階を区切って出します。一度に全部を頼むと、どの設定でどう変わったのかが追えなくなります。例えば、次のような依頼になります。

infra/poison-lab/ に、シャード1本のKinesisストリームと、
JSONの poison が true のレコードで例外を出すPython関数を
Terraformで作ってください。イベントソースマッピングは
batch_size=10、再試行と最大経過時間は指定しないでください。
terraform plan までで止めて、差分を見せてください。

planの差分を読み、イベントソースマッピングに余計な引数が入っていないことを確かめてから、人がapplyを承認します。次に、毒レコードと正常なレコードを順に送ります。

aws kinesis put-record --stream-name poison-lab \
  --partition-key k1 --cli-binary-format raw-in-base64-out \
  --data '{"poison": true}'
 
aws kinesis put-record --stream-name poison-lab \
  --partition-key k1 --cli-binary-format raw-in-base64-out \
  --data '{"poison": false, "id": 1}'

送った後に見るのは、関数のログと、LambdaのIteratorAgeメトリクスです。IteratorAgeは、ストリームがレコードを受け取ってから、イベントソースマッピングが関数に送るまでの経過時間を示します。段階1では、この値が伸び続けるはずです。シャードが進まないことの、数字での確認になります。

CloudWatch Logs Insightsでエラーの急増を調べるクエリをClaude Codeに書かせれば、ログからの原因の切り分けも同じ流れで進められます。

CLAUDE.mdに置く規約の例

毎回の依頼で同じ注意を書かなくて済むよう、プロジェクトのCLAUDE.mdに前提を置きます。

## 検証用AWS環境のルール
- 操作してよいのは AWS_PROFILE=sandbox のときだけ
- terraform apply と destroy は人が実行する。plan までで止める
- Kinesis の再現では、段階ごとに設定を1つだけ変える
- 失敗先のバケットを消すときは、先に destination_config を外す

最後の1行は、S3を失敗先にした場合の対策です。バケットだけを先に消すと、設定が宛先を指したまま残る状態になります。

本番に移すときに、見直しておく設定

再現でわかるのは、既定値が安全側ではないことです。本番の設定を決めるときは、次の観点で見直します。

• MaximumRecordAgeInSecondsは、その業務で許せる遅延より短くする
• MaximumRetryAttemptsは、一時的な障害を吸収できる最小の回数にする
• 失敗先は、レコード本体を残したいかどうかでSQSかS3を選ぶ
• 関数は冪等にする(Lambdaは各レコードを少なくとも1回処理するため、重複が起きうる)

シャードを増やすのも、影響を狭める方法です。ドキュメントによれば、シャードが多いほど同時に処理されるバッチが増え、エラーの影響が小さくなります。ParallelizationFactorを使えば、1シャードを複数の呼び出しで同時に処理できます。既定は1、最大は10です。

関数の設計やデプロイを広く任せたい場合は、aws-serverlessプラグインでLambdaを設計から検証まで進める手順が入口になります。

まとめ

既定のKinesis連携は、毒レコード1件でシャードが止まり、Lambda側に諦める条件がありません。再試行と最大経過時間に上限を置き、失敗先を決め、バッチを割る設定を重ねると、止まる範囲と時間を絞れます。

Claude Codeに任せるのは、Terraformの記述とplanの差分の読み上げまでです。applyと本番への反映は人が握り、サンドボックス以外では動かさないフックを先に置いておくのが、この再現を安全に繰り返す条件になります。