Polling topik Kafka dalam mode Antrian

Mode antrian adalah mode konsumsi yang menggunakan grup berbagi, yang tersedia di Apache Kafka 4.0 (Kafka Queue Poller memerlukan versi 4.2 atau yang lebih baru). Dengan mode Antrian, Anda dapat memproses pesan topik Kafka dengan cara throughput tinggi yang tidak teratur menggunakan grup berbagi. Tidak seperti kelompok konsumen Kafka tradisional yang menetapkan partisi secara eksklusif untuk konsumen, grup berbagi memungkinkan beberapa konsumen membaca dari partisi yang sama secara bersamaan. Ini mirip dengan cara Amazon SQS memproses pesan dari antrian.

Saat Anda membuat pemetaan sumber peristiwa untuk pengelolaan sendiri Apache Kafka cluster, Anda dapat memilih antara dua mode konsumsi:

• Streaming (default). Menggunakan kelompok konsumen Kafka dengan pemesanan per partisi. Lambda menetapkan setiap partisi ke satu konsumen.

• Antrian. Menggunakan grup berbagi, yang tersedia di Apache Kafka 4.0, untuk pemrosesan throughput tinggi yang tidak teratur (Kafka Queue Poller memerlukan versi 4.2 atau lebih baru). Dengan grup berbagi, beberapa konsumen dapat membaca dari partisi yang sama secara bersamaan, mirip dengan cara Amazon SQS memproses pesan dari antrian.

Jika Anda tidak menentukan mode konsumsi, pemetaan sumber peristiwa menggunakan Stream. Halaman ini mencakup mode konsumsi antrian. Untuk informasi umum tentang menggunakan Lambda dengan self-managed Apache Kafka, lihat Menggunakan Lambda dengan Apache Kafka yang dikelola sendiri.

Dalam mode Antrian, Lambda menggunakan Kafka Share Consumer untuk polling pesan dari topik. Setiap poller peristiwa memproses pesan di semua partisi tanpa pemesanan tingkat partisi. Ini meningkatkan konkurensi untuk setiap poller.

Persyaratan minimum

Untuk menggunakan mode konsumsi antrian, cluster Kafka Anda harus memenuhi persyaratan berikut:

• Apache Kafka versi 4.2 atau lebih baru dengan grup berbagi diaktifkan. Kafka memungkinkan share.version secara default, dan Anda dapat mengelolanya melalui kafka-features.sh.

• Self-managed Apache Kafka (SMK) cluster saja. Streaming Terkelola Amazon untuk Apache Kafka (Amazon MSK) saat ini tidak mendukung mode Antrian. AWS menambahkan dukungan MSK setelah MSK menawarkan Kafka 4.2 atau lebih baru dengan grup berbagi diaktifkan.

Perbandingan Streaming dan Antrian

Fitur	Streaming (default)	Antrean
Pemesanan	Per-partition pemesanan dijamin	Tidak ada jaminan pemesanan
Model konsumen	Kelompok konsumen (penugasan partisi eksklusif)	Bagikan grup (akses partisi bersamaan)
Throughput per event poller	Hingga 5 MB/s	Hingga 5 MB/s
Pemanggilan bersamaan per poller peristiwa	Hingga 5	Hingga 10
Mode penskalaan	On-demand dan disediakan	On-demand dan disediakan
Berbagi EPU	Dapat berbagi EPU dengan pemetaan sumber acara Kafka Streams lainnya	Terisolasi. Pemetaan sumber peristiwa antrian hanya dibagikan dengan pemetaan sumber peristiwa Antrian lainnya
Kasus penggunaan	Pemrosesan acara yang dipesan	High-throughput beban kerja yang tidak teratur

Buat pemetaan antrian

Untuk membuat pemetaan sumber peristiwa mode Antrian, atur ConsumptionMode bidang SelfManagedKafkaEventSourceConfig kepada Queue. Dalam contoh berikut, ganti my-function, titik akhir broker, dan my-queue-topic dengan nilai-nilai Anda sendiri.

AWS CLI

Berikut ini AWS Perintah CLI membuat pemetaan sumber peristiwa mode antrian. Ganti my-function dengan nama fungsi Anda, titik akhir broker dengan broker Anda, dan my-queue-topic dengan topik Anda.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 1, "MaximumPollers": 10}'

AWS CloudFormation

Berikut ini AWS CloudFormation template membuat pemetaan sumber peristiwa mode antrian. Ganti titik akhir broker dan my-queue-topic dengan nilai-nilai Anda sendiri.

MyKafkaQueueESM:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    FunctionName: !Ref MyFunction
    SelfManagedEventSource:
      Endpoints:
        KafkaBootstrapServers:
          - broker1.us-west-2.example.com:9092
          - broker2.us-west-2.example.com:9092
    Topics:
      - my-queue-topic
    BatchSize: 100
    SelfManagedKafkaEventSourceConfig:
      ConsumptionMode: Queue
    ProvisionedPollerConfig:
      MinimumPollers: 1
      MaximumPollers: 10

Parameter konfigurasi

Tabel berikut mencantumkan parameter yang berlaku untuk pemetaan sumber peristiwa mode antrian.

Parameter	Deskripsi	Rentang/Default
BatchSize	Jumlah maksimum catatan per batch yang dikirim ke fungsi	1—10.000. Default: 100
Enabled	Apakah pemetaan sumber peristiwa aktif	true atau false. Default: true
FilterCriteria	Pola pemfilteran peristiwa untuk memproses hanya catatan yang cocok	Hingga 5 pola filter
MaximumRetryAttempts	Percobaan ulang maksimum untuk catatan yang gagal	-1-10.000. Default: 0 (dalam mode Antrian, broker Kafka melakukan percobaan ulang ketika tidak ada tujuan pada kegagalan yang dikonfigurasi)
BisectBatchOnFunctionError	Pisahkan batch menjadi dua pada kesalahan fungsi	true atau false. Default: false
ProvisionedPollerConfig.MinimumPollers	Poller acara minimum (mode disediakan)	1—200. Default: 1
ProvisionedPollerConfig.MaximumPollers	Poller peristiwa maksimum (mode disediakan)	1—2.000. Standar: 200
ProvisionedPollerConfig.PollerGroupName	Nama grup untuk berbagi EPU (mode disediakan, hanya pemetaan sumber acara antrian)	String. Hingga 100 pemetaan sumber peristiwa per grup
SelfManagedKafkaEventSourceConfig.ConsumptionMode	Mode konsumsi untuk sumber acara	Harus Queue untuk mode Antrian
SchemaRegistryConfig	Registri skema untuk validasi acara	Lihat Menggunakan pendaftar skema dengan sumber acara Kafka di Lambda

penting

MaximumBatchingWindowInSeconds Parameter adalah tidak didukung untuk pemetaan sumber peristiwa mode antrian. Konsumen Saham Kafka menggunakan pengakuan eksplisit. Semua catatan dalam batch harus diakui secara eksplisit sebelum konsumen dapat mengambil yang baru. Ini tidak kompatibel dengan jendela batching berbasis waktu. Jika Anda menentukan MaximumBatchingWindowInSeconds saat membuat pemetaan sumber peristiwa mode Antrian, API mengembalikan kesalahan validasi.

Fitur pemetaan sumber acara yang didukung

Tabel berikut mencantumkan fitur pemetaan sumber peristiwa Lambda yang tersedia dengan mode Antrian.

Fitur	Didukung	Catatan
Mode yang disediakan	Ya	Konfigurasikan MinimumPollers dan MaximumPollers
On-demand mode	Ya	Mode penskalaan default. Lambda menskalakan poller secara otomatis hingga 5 MB/s atau 50 pemanggilan bersamaan, mana yang dicapai lebih dulu. Tidak diperlukan konfigurasi tambahan.
Pemfilteran peristiwa	Ya	Hingga 5 pola filter per pemetaan sumber peristiwa
Skema registri	Ya	Avro dan Protobuf dengan Glue Schema Registry atau Confluent Schema Registry. Pemfilteran registri skema hanya tersedia dalam mode yang disediakan.
Low-latency pengolahan	Ya	Tersedia dalam mode disediakan
Coba lagi konfigurasi	Ya	MaximumRetryAttempts, BisectBatchOnFunctionError
On-failure tujuan (topik Kafka)	Ya	Mengirim catatan yang gagal ke topik Kafka
On-failure tujuan (SNS/SQS)	Ya	Kirim catatan yang gagal ke SNS atau SQS
Pencatatan pemetaan sumber peristiwa	Ya	CloudWatch Log untuk diagnostik pemetaan sumber peristiwa
Pengelompokan pemetaan sumber acara (PollerGroupName)	Ya	Pemetaan sumber peristiwa mode antrian dapat berbagi EPU dengan pemetaan sumber peristiwa mode antrian lainnya saja (mode disediakan)
Jendela batch	Tidak	Tidak didukung karena perilaku mode pemblokiran Share Consumer
Posisi awal	Tidak	StartingPosition dan StartingPositionTimestamp tidak didukung. Mode antrian selalu dibaca dari posisi berkomitmen grup berbagi.

Mode penskalaan

Pemetaan sumber peristiwa mode antrian mendukung mode penskalaan sesuai permintaan (default) dan yang disediakan.

On-demand mode (default)

Dalam mode on-demand, Lambda secara otomatis mengelola poller peristiwa untuk pemetaan sumber peristiwa mode Antrian Anda.

On-demand mode menskalakan hingga 5 poller peristiwa per pemetaan sumber peristiwa. Mencapai plafonnya di 5 MB/s throughput agregat atau 50 pemanggilan bersamaan (5 poller dikalikan 10). Ini terjadi batas mana pun yang tercapai lebih dulu. Untuk menskalakan melampaui batas ini, gunakan mode yang disediakan. On-demand mode tidak memerlukan konfigurasi tambahan di luar parameter pemetaan sumber peristiwa standar.

Tabel berikut mencantumkan batas mode sesuai permintaan.

Kuota	Nilai
Throughput per event poller	Hingga 1 MB/s
Pemanggilan bersamaan per poller peristiwa	Hingga 10
Poller acara maksimum	5
Throughput agregat maksimum	Hingga 5 MB/s (5 poller kali 1 MB/s)
Konkurensi agregat maksimum	Hingga 50 pemanggilan bersamaan (5 poller kali 10)

Contoh - Pemetaan sumber peristiwa mode antrian dalam mode on-demand (AWS CLI)

Berikut ini AWS Perintah CLI membuat pemetaan sumber peristiwa mode Antrian dalam mode on-demand. Ganti my-function, titik akhir broker, my-queue-topic, ID subnet, dan ID grup keamanan dengan nilai Anda sendiri.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Mode yang disediakan

Untuk beban kerja dengan persyaratan kinerja yang ketat atau pola lalu lintas yang runcing, gunakan mode yang disediakan. Mode yang disediakan memberi Anda kontrol langsung atas jumlah poller peristiwa minimum dan maksimum yang didedikasikan untuk pemetaan sumber peristiwa mode Antrian Anda.

Setiap poller peristiwa mode Antrian dalam mode disediakan mendukung hingga 5 MB/s throughput dan hingga 10 pemanggilan Lambda bersamaan.

Tabel berikut mencantumkan batas mode yang disediakan.

Kuota	Nilai
Throughput per event poller	Hingga 5 MB/s
Pemanggilan bersamaan per poller peristiwa	Hingga 10
Poller acara minimum (MinimumPollers)	1—200. Default: 1
Poller acara maksimum (MaximumPollers)	1—2.000. Standar: 200
Pemetaan sumber peristiwa per grup poller	Hingga 100
Agregat poller maksimum per kelompok poller	2.000
Throughput per EPU (agregat)	Hingga 5 MB/s per poller; agregat tergantung pada kepadatan kemasan poller

Mode yang disediakan dikenakan biaya berdasarkan penggunaan Event Poller Unit (EPU). Untuk detail harga, lihat AWS Lambda Harga.

Contoh - Pemetaan sumber peristiwa mode antrian dengan mode disediakan (AWS CLI)

Berikut ini AWS Perintah CLI membuat pemetaan sumber peristiwa mode antrian dalam mode disediakan.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Ganti placeholder berikut dengan nilai Anda sendiri:

• subnet-0123456789abcdef0 dan subnet-0123456789abcdef1: ID subnet Anda.

• sg-0123456789abcdef0: ID grup keamanan Anda.

• MinimumPollers dan MaximumPollers: diatur sesuai dengan kebutuhan throughput Anda.

catatan

Untuk beralih dari mode yang disediakan kembali ke mode sesuai permintaan, berikan tanda kosong ProvisionedPollerConfig:

PROMPT> aws lambda update-event-source-mapping \
  --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
  --provisioned-poller-config '{}'

Pemilihan mode penskalaan

Pertimbangan	On-demand	Disediakan
Pengaturan	Tidak ada konfigurasi tambahan	Konfigurasikan MinimumPollers dan MaximumPollers
Biaya	Tidak ada biaya EPU tambahan	Biaya EPU berdasarkan penggunaan per jam
Terbaik untuk	Low-concurrency beban kerja	Persyaratan latensi yang ketat, lalu lintas runcing
Pengelompokan pemetaan sumber acara (PollerGroupName)	Tidak didukung	Didukung. Bagikan EPU di seluruh pemetaan sumber peristiwa Antrian
Low-latency pengolahan	Tidak tersedia	Available
Penyaringan registri skema (Avro/Protobuf)	Tidak tersedia	Available

Isolasi berbagi EPU

Dalam mode yang disediakan, Anda dapat mengelompokkan beberapa pemetaan sumber peristiwa dalam VPC yang sama untuk berbagi kapasitas Event Poller Unit (EPU) dengan menetapkan kapasitas yang sama PollerGroupName. Pengelompokan pemetaan sumber peristiwa memungkinkan pemetaan sumber peristiwa throughput rendah dikemas ke EPU bersama alih-alih masing-masing memerlukan kapasitas khusus, mengurangi biaya mode disediakan. Lihat perinciannya di Optimalisasi Biaya untuk mode Provisioned.

Pemetaan sumber peristiwa mode antrian hanya berbagi EPU dengan pemetaan sumber peristiwa mode Antrian lainnya. SEBUAH PollerGroupName digunakan oleh pemetaan sumber peristiwa mode Antrian tidak dapat dibagikan dengan pemetaan sumber peristiwa Kafka Streams, dan sebaliknya. Jika Anda mengkonfigurasi a PollerGroupName sudah digunakan oleh mode konsumsi lainnya, CreateEventSourceMapping mengembalikan kesalahan validasi. Untuk menggunakan pengelompokan pemetaan sumber peristiwa dengan mode Antrian, buat PollerGroupName didedikasikan untuk pemetaan sumber acara Antrian Anda.

Contoh peristiwa

Ketika Lambda memanggil fungsi Anda dengan sumber peristiwa mode Antrian, payload peristiwa identik dengan peristiwa Kafka standar. Setiap catatan mencakup topik, partisi, offset, stempel waktu, dan kunci dan nilai yang dikodekan base64:

{
  "eventSource": "SelfManagedKafka",
  "bootstrapServers": "broker1.example.com:9092,broker2.example.com:9092",
  "records": {
    "my-queue-topic-0": [
      {
        "topic": "my-queue-topic",
        "partition": 0,
        "offset": 15,
        "timestamp": 1545084650987,
        "timestampType": "CREATE_TIME",
        "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
        "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
        "headers": [
          {
            "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
          }
        ]
      }
    ],
    "my-queue-topic-3": [
      {
        "topic": "my-queue-topic",
        "partition": 3,
        "offset": 42,
        "timestamp": 1545084651000,
        "timestampType": "CREATE_TIME",
        "key": "YW5vdGhlcktleQ==",
        "value": "VGhpcyBpcyBhbm90aGVyIG1lc3NhZ2U=",
        "headers": []
      }
    ]
  }
}

Idempotensi diperlukan

Pemetaan sumber peristiwa lambda memproses setiap peristiwa setidaknya sekali, sehingga pemrosesan rekaman duplikat dapat terjadi. Karena mode Antrian tidak menjamin pemesanan, pemrosesan duplikat atau di luar urutan dapat menghasilkan hasil yang salah, seperti penghitungan ganda atau status tidak konsisten, jika kode fungsi Anda tidak berpotensi. Jadikan kode fungsi Anda idempoten.

Batasan

• Self-managed Kelompok-kelompok Kafka saja. Amazon MSK tidak didukung saat peluncuran.

• Apache Kafka 4.2 atau yang lebih baru diperlukan. Cluster harus mendukung protokol Share Consumer dengan grup berbagi diaktifkan.

• Jendela batching tidak didukung. MaximumBatchingWindowInSeconds parameter tidak dapat digunakan dengan pemetaan sumber peristiwa Kafka Queue.

• Posisi awal tidak didukung. StartingPosition dan StartingPositionTimestamp tidak dapat digunakan dengan pemetaan sumber peristiwa mode Antrian. Grup saham mengkonsumsi dari posisi offset yang dikomitmennya.

• Isolasi EPU. Pemetaan sumber peristiwa mode antrian tidak dapat berbagi EPU dengan pemetaan sumber peristiwa Kafka Streams. Gunakan terpisah PollerGroupName nilai-nilai.