Kafka トピックを障害発生時の送信先として使用する

Kafka トピックは、Kafka イベントソースマッピングの障害発生時の送信先として設定できます。再試行回数を超えた後、またはレコードが最大経過時間を超えた後に Lambda がレコードを処理できなくなると、Lambda は失敗したレコードを指定された Kafka トピックに送信して、後で処理します。

Kafka 障害発生時の送信先のしくみ

Kafka トピックを障害発生時の送信先として設定すると、Lambda は Kafka プロデューサーとして機能し、失敗したレコードを送信先トピックに書き込みます。これにより、Kafka インフラストラクチャ内にデッドレタートピック (DLT) のパターンが作成されます。

Kafka 障害発生時の送信先の設定

Kafka イベントソースマッピングを作成または更新するときに、Kafka トピックを障害発生時の送信先として設定できます。

Kafka 送信先の設定 (コンソール)

Kafka 送信先の設定 (AWS CLI)

kafka:// プレフィックスを使用して、障害発生時の送信先として Kafka トピックを指定します。

Kafka 送信先を使用したイベントソースマッピングの作成

次の例では、障害発生時の送信先として Kafka トピックを使用して Amazon MSK イベントソースマッピングを作成します。

aws lambda create-event-source-mapping \
  --function-name my-kafka-function \
  --topics AWSKafkaTopic \
  --event-source-arn arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abc123 \
  --starting-position LATEST \
  --provisioned-poller-config MinimumPollers=1,MaximumPollers=3 \
  --destination-config '{"OnFailure":{"Destination":"kafka://failed-records-topic"}}'

セルフマネージド型の Kafka の場合は、同じ構文を使用します。

aws lambda create-event-source-mapping \
  --function-name my-kafka-function \
  --topics AWSKafkaTopic \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc.xyz.com:9092"]}}' \
  --starting-position LATEST \
  --provisioned-poller-config MinimumPollers=1,MaximumPollers=3 \
  --destination-config '{"OnFailure":{"Destination":"kafka://failed-records-topic"}}'

Kafka 送信先の更新

update-event-source-mapping コマンドを使用して Kafka 送信先を追加または変更します。

aws lambda update-event-source-mapping \
  --uuid 12345678-1234-1234-1234-123456789012 \
  --destination-config '{"OnFailure":{"Destination":"kafka://failed-records-topic"}}'

Kafka 送信先のレコード形式

Lambda が失敗したレコードを Kafka トピックに送信すると、各メッセージにはその障害に関するメタデータと実際のレコードコンテンツの両方が含まれます。

障害のメタデータ

メタデータには、レコードが失敗した理由に関する情報と元のバッチに関する詳細が含まれます。

{
  "requestContext": {
    "requestId": "e4b46cbf-b738-xmpl-8880-a18cdf61200e",
    "functionArn": "arn:aws:lambda:us-east-1:123456789012:function:my-function:$LATEST",
    "condition": "RetriesExhausted",
    "approximateInvokeCount": 3
  },
  "responseContext": {
    "statusCode": 200,
    "executedVersion": "$LATEST",
    "functionError": "Unhandled"
  },
  "version": "1.0",
  "timestamp": "2019-11-14T18:16:05.568Z",
  "KafkaBatchInfo": {
    "batchSize": 1,
    "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abc123",
    "bootstrapServers": "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com:9098",
    "payloadSize": 1162,
    "recordInfo": {
      "offset": "49601189658422359378836298521827638475320189012309704722",
      "timestamp": "2019-11-14T18:16:04.835Z"
    }
  },
  "payload": {
    "bootstrapServers": "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com:9098",
    "eventSource": "aws:kafka",
    "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abc123",
    "records": {
      "my-topic-0": [
        {
          "headers": [],
          "key": "dGVzdC1rZXk=",
          "offset": 100,
          "partition": 0,
          "timestamp": 1749116692330,
          "timestampType": "CREATE_TIME",
          "topic": "my-topic",
          "value": "dGVzdC12YWx1ZQ=="
        }
      ]
    }
  }
}

パーティションキーの動作

Lambda は、送信先トピックに生成するときに、元のレコードと同じパーティションキーを使用します。元のレコードにキーがない場合、Lambda は送信先トピックで使用可能なすべてのパーティションで Kafka のデフォルトのラウンドロビンパーティショニングを使用します。

要件と制限