Menangkap batch yang dibuang untuk sumber acara Apache Kafka yang dikelola sendiri

Untuk menyimpan catatan pemanggilan pemetaan sumber peristiwa yang gagal, tambahkan tujuan ke pemetaan sumber peristiwa fungsi Anda. Setiap catatan yang dikirim ke tujuan adalah JSON dokumen dengan metadata tentang pemanggilan yang gagal. Anda dapat mengonfigurasi SNS topik Amazon, SQS antrian Amazon, atau bucket S3 sebagai tujuan. Peran eksekusi Anda harus memiliki izin untuk tujuan:

• Untuk SQS tujuan: sqs: SendMessage

• Untuk SNS tujuan: SNS: Publikasikan

• Untuk tujuan bucket S3: s3: PutObject dan s3: ListBuckets

Anda harus menerapkan VPC endpoint untuk layanan tujuan on-failure Anda di dalam cluster Apache Kafka Anda. VPC

Selain itu, jika Anda mengonfigurasi KMS kunci tujuan, Lambda memerlukan izin berikut tergantung pada jenis tujuan:

• Jika Anda telah mengaktifkan enkripsi dengan KMS kunci Anda sendiri untuk tujuan S3, kms: GenerateDataKey diperlukan. Jika tujuan bucket KMS kunci dan S3 berada di akun yang berbeda dari fungsi Lambda dan peran eksekusi, konfigurasikan KMS kunci untuk mempercayai peran eksekusi agar memungkinkan kms:. GenerateDataKey

• Jika Anda telah mengaktifkan enkripsi dengan KMS kunci Anda sendiri untuk SQS tujuan, KMS: Decrypt dan kms: diperlukan. GenerateDataKey Jika tujuan KMS kunci dan SQS antrean berada di akun yang berbeda dari fungsi Lambda dan peran eksekusi, konfigurasikan KMS kunci untuk mempercayai peran eksekusi untuk mengizinkan KMS:Decrypt, kms:, kms:, dan kmsGenerateDataKey:. DescribeKey ReEncrypt

• Jika Anda telah mengaktifkan enkripsi dengan KMS kunci Anda sendiri untuk SNS tujuan, KMS: Decrypt dan kms: diperlukan. GenerateDataKey Jika tujuan KMS kunci dan SNS topik berada di akun yang berbeda dari fungsi Lambda dan peran eksekusi, konfigurasikan KMS kunci untuk mempercayai peran eksekusi untuk mengizinkan KMS:Decrypt, kms:, kms:, dan kms:GenerateDataKey. DescribeKey ReEncrypt

Mengonfigurasi tujuan yang gagal untuk pemetaan sumber acara Apache Kafka yang dikelola sendiri

Untuk mengonfigurasi tujuan saat gagal menggunakan konsol, ikuti langkah-langkah berikut:

1. Buka halaman Fungsi di konsol Lambda.

2. Pilih fungsi.

3. Di bagian Gambaran umum fungsi, pilih Tambahkan tujuan.

4. Untuk Sumber, pilih Pemanggilan pemetaan sumber acara.

5. Untuk pemetaan sumber peristiwa, pilih sumber peristiwa yang dikonfigurasi untuk fungsi ini.

6. Untuk Kondisi, pilih On failure. Untuk pemanggilan pemetaan sumber peristiwa, ini adalah satu-satunya kondisi yang diterima.

7. Untuk tipe Tujuan, pilih tipe tujuan yang Lambda kirimkan catatan pemanggilan.

8. Untuk Tujuan, pilih sumber daya.

9. Pilih Simpan.

Anda juga dapat mengonfigurasi tujuan pada kegagalan menggunakan file. AWS CLI Misalnya, create-event-source-mappingperintah berikut menambahkan pemetaan sumber peristiwa dengan tujuan SQS kegagalan ke: MyFunction

aws lambda create-event-source-mapping \
--function-name "MyFunction" \
--event-source-arn arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2 \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-1:123456789012:dest-queue"}}'

update-event-source-mappingPerintah berikut menambahkan tujuan kegagalan S3 ke sumber acara yang terkait dengan input: uuid

aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": "arn:aws:s3:::dest-bucket"}}'

Untuk menghapus tujuan, berikan string kosong sebagai argumen ke destination-config parameter:

aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": ""}}'

SNSdan SQS contoh catatan pemanggilan

Contoh berikut menunjukkan apa yang Lambda kirim ke SNS topik atau tujuan SQS antrian untuk pemanggilan sumber acara Kafka yang gagal. Setiap kunci di bawah recordsInfo berisi topik dan partisi Kafka, dipisahkan oleh tanda hubung. Misalnya, untuk kuncinya"Topic-0", Topic adalah topik Kafka, dan 0 merupakan partisi. Untuk setiap topik dan partisi, Anda dapat menggunakan data offset dan stempel waktu untuk menemukan catatan pemanggilan asli.

{
    "requestContext": {
        "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
        "functionArn": "arn:aws:lambda:us-east-1:123456789012:function:myfunction",
        "condition": "RetryAttemptsExhausted" | "MaximumPayloadSizeExceeded",
        "approximateInvokeCount": 1
    },
    "responseContext": { // null if record is MaximumPayloadSizeExceeded
        "statusCode": 200,
        "executedVersion": "$LATEST",
        "functionError": "Unhandled"
    },
    "version": "1.0",
    "timestamp": "2019-11-14T00:38:06.021Z",
    "KafkaBatchInfo": {
        "batchSize": 500,
        "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
        "bootstrapServers": "...",
        "payloadSize": 2039086, // In bytes
        "recordsInfo": {
            "Topic-0": {
                "firstRecordOffset": "49601189658422359378836298521827638475320189012309704722",
                "lastRecordOffset": "49601189658422359378836298522902373528957594348623495186",
                "firstRecordTimestamp": "2019-11-14T00:38:04.835Z",
                "lastRecordTimestamp": "2019-11-14T00:38:05.580Z",
            },
            "Topic-1": {
                "firstRecordOffset": "49601189658422359378836298521827638475320189012309704722",
                "lastRecordOffset": "49601189658422359378836298522902373528957594348623495186",
                "firstRecordTimestamp": "2019-11-14T00:38:04.835Z",
                "lastRecordTimestamp": "2019-11-14T00:38:05.580Z",
            }
        }
    }
}

Catatan pemanggilan contoh tujuan S3

Untuk tujuan S3, Lambda mengirimkan seluruh catatan pemanggilan bersama dengan metadata ke tujuan. Contoh berikut menunjukkan bahwa Lambda mengirim ke tujuan bucket S3 untuk pemanggilan sumber peristiwa Kafka yang gagal. Selain semua bidang dari contoh sebelumnya untuk SQS dan SNS tujuan, payload bidang berisi catatan pemanggilan asli sebagai string yang JSON diloloskan.

{
    "requestContext": {
        "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
        "functionArn": "arn:aws:lambda:us-east-1:123456789012:function:myfunction",
        "condition": "RetryAttemptsExhausted" | "MaximumPayloadSizeExceeded",
        "approximateInvokeCount": 1
    },
    "responseContext": { // null if record is MaximumPayloadSizeExceeded
        "statusCode": 200,
        "executedVersion": "$LATEST",
        "functionError": "Unhandled"
    },
    "version": "1.0",
    "timestamp": "2019-11-14T00:38:06.021Z",
    "KafkaBatchInfo": {
        "batchSize": 500,
        "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
        "bootstrapServers": "...",
        "payloadSize": 2039086, // In bytes
        "recordsInfo": {
            "Topic-0": {
                "firstRecordOffset": "49601189658422359378836298521827638475320189012309704722",
                "lastRecordOffset": "49601189658422359378836298522902373528957594348623495186",
                "firstRecordTimestamp": "2019-11-14T00:38:04.835Z",
                "lastRecordTimestamp": "2019-11-14T00:38:05.580Z",
            },
            "Topic-1": {
                "firstRecordOffset": "49601189658422359378836298521827638475320189012309704722",
                "lastRecordOffset": "49601189658422359378836298522902373528957594348623495186",
                "firstRecordTimestamp": "2019-11-14T00:38:04.835Z",
                "lastRecordTimestamp": "2019-11-14T00:38:05.580Z",
            }
        }
    },
    "payload": "<Whole Event>" // Only available in S3
}

Tip

Sebaiknya aktifkan versi S3 di bucket tujuan Anda.