Firehoseの変換Lambda、異常レコードもrecordIdは返さないとエラーになる

  • AWS
  • Firehose
  • Lambda

この記事は、構成と文章の整理にAIを使っています。
あくまで読みやすくするための補助で、内容そのものをAIにすべて任せて書いているわけではありませんが、ところどころ相談しながら整えています。

Kinesis Data Firehoseのデータ変換でLambdaを使っていて、正常なレコードだけreturnして異常なレコードを除外したらOne or more record Ids were not returned. Ensure that the Lambda function returns all received record Ids.というエラーが出た。

何が起きたか

Firehoseの変換Lambdaは、受け取ったevent.recordsに含まれる全レコードについて、対応するrecordIdをレスポンスに含める必要がある。正常/異常に関わらず、Lambdaに渡されたレコード数とレスポンスのレコード数が一致していないとこのエラーになる。異常なレコードを単に配列から除外してreturnする、という実装は不可。

対応

各レコードに対してrecordIdとresultを必ず設定して返す。resultに指定できる値は3つ。

  • Ok: 変換成功。data（base64エンコード済み）を返す
  • Dropped: 意図的に破棄
  • ProcessingFailed: 変換失敗
import { FirehoseTransformationEvent, FirehoseTransformationResult } from 'aws-lambda';


export const handler = async (
  event: FirehoseTransformationEvent
): Promise<FirehoseTransformationResult> => {
  const records = event.records.map((record) => {
    try {
      const transformed = transform(record.data);
      return { recordId: record.recordId, result: 'Ok', data: transformed };
    } catch {
      // 変換不能なレコード。recordIdは必ず含める
      return { recordId: record.recordId, result: 'ProcessingFailed' };
    }
  });


  return { records };
};

OkとDropped、ProcessingFailedの違い

3つとも「recordIdを返す」という点では同じだが、Firehose内部での扱いが異なる。

  • Ok・Dropped → Firehose的には「処理成功」扱い
  • ProcessingFailed → 「処理失敗」扱い

この違いはエラー出力先の挙動に直結する。ProcessingFailedにしたレコードは、設定していればS3のエラー出力先（ProcessingFailed用のプレフィックス）に送られる。一方Droppedは成功扱いのため、エラー出力先には送られず単に捨てられるだけになる。

なので使い分けの基準は以下になる。

  • 異常データとして後から追跡・調査したい → ProcessingFailed
  • ノイズとして完全に無視してよく、追跡不要 → Dropped

resultレスポンスに指定できるフィールドはrecordId / result / dataの3つのみで、エラーメッセージ専用のフィールドは存在しない。異常内容を残したい場合はdataに詰めるか、Lambda側でCloudWatch Logsに出力する形になる。