xkafka — Cara lebih mudah menggunakan Kafka di Go

• xkafka adalah library open-source yang memungkinkan Kafka digunakan sesederhana layanan HTTP di lingkungan Go
• Saat menggunakan confluent-kafka-go, biasanya dibutuhkan loop pemrosesan yang kompleks dan banyak kode boilerplate, tetapi xkafka memungkinkan Anda fokus pada logika inti dengan struktur Handler, Middleware, Message
• Publikasi dan konsumsi pesan ditangani secara intuitif seperti pola permintaan/respons HTTP, sekaligus banyak menyembunyikan kompleksitas Kafka seperti manajemen offset, pengaturan konkurensi, dan penanganan error
• Mendukung dengan sederhana berbagai pola yang dibutuhkan di layanan produksi, seperti pemrosesan Streaming/Batch, pemrosesan sekuensial/asinkron, serta jaminan At-most-once/At-least-once
• Pola yang dibutuhkan dalam praktik seperti penanganan error bertingkat dan retry/logging/metrics berbasis middleware juga dapat diterapkan dengan mudah

---

Kafka bergaya HTTP

• xkafka adalah library yang mengabstraksikan Kafka di Go layaknya layanan HTTP
  • Message mirip dengan permintaan HTTP, mencakup topik/partisi/offset/key/value/header/callback dan lainnya
  • Handler memproses logika bisnis seperti HTTP Handler
  • Middleware memungkinkan fitur tambahan seperti logging, metrics, dan retry diterapkan terpisah dari logika bisnis

Publikasi pesan (Publishing Messages)

• Setelah membuat Producer dengan xkafka.NewProducer, buat objek pesan lalu publikasikan dengan fungsi Publish
• Mendukung publikasi asinkron (AsyncPublish) dan pendaftaran callback, sehingga pemrosesan event asinkron atau kebutuhan performa tinggi menjadi lebih mudah
• Pengiriman pesan diproses di goroutine latar belakang, dan status pengiriman dapat dilacak melalui callback

Konsumsi pesan (Consuming Messages)

• Saat membuat Consumer, tentukan fungsi Handler serta topik/broker/konfigurasi dan lainnya
• Middleware dapat ditambahkan dengan consumer.Use()
• Konsumsi pesan dimulai dengan consumer.Run(ctx)

Streaming vs. Batch

• Streaming: setiap pesan diproses langsung satu per satu saat tiba. Cocok untuk throughput rendah, penghematan memori, atau jaminan pemrosesan yang kuat
• Batch: pesan diproses dalam kelompok berdasarkan jumlah tertentu atau interval waktu. Cocok untuk sistem throughput tinggi atau mengurangi beban downstream

Sekuensial atau Async

• Default-nya adalah pemrosesan sekuensial (Sequential) — pesan berikutnya baru dibaca setelah satu pesan selesai diproses
• Saat menggunakan xkafka.Concurrency(N), tersedia mode asinkron (Async) yang memproses N pesan (atau batch) secara bersamaan

Manajemen offset

• Perilaku default Kafka adalah memajukan offset segera setelah pesan diterima, sehingga ada kemungkinan kehilangan pesan saat terjadi gangguan
• xkafka mengatur enable.auto.offset.store=false, sehingga offset hanya disimpan setelah pesan (atau batch) selesai diproses
• Tanpa perlu mengelola status pesan di DB atau queue terpisah, jaminan pemrosesan tetap bisa diperoleh di Kafka

• Jaminan At-Most-Once

  • Secara default, offset di-commit di latar belakang sesuai enable.auto.commit=true milik Kafka
  • Dengan xkafka.ManualCommit(true) dan pemrosesan sekuensial, offset di-commit sebelum membaca setiap pesan/batch untuk menjamin At-most-once

• Jaminan At-Least-Once

  • Dengan menggabungkan xkafka.ManualCommit(true) dan konkurensi (N>1), offset dapat di-commit secara sinkron dan berurutan bahkan saat pemrosesan paralel berlangsung
  • Pola jaminan At-least-once dapat diterapkan dengan mudah

Penanganan error

• Level Handler

  • Di dalam Handler, error aplikasi dapat ditangani dan pesan bisa dikirim ke Dead Letter Queue
  • Kontrol dilakukan secara eksplisit: msg.AckSuccess() saat berhasil, msg.AckSkip() saat dilewati, msg.AckFail(err) saat gagal

• Level Middleware

  • Logika umum seperti retry dan logging error dapat digunakan ulang di banyak Handler melalui middleware
  • Kebijakan retry atau metode penanganan yang berbeda dapat diterapkan dengan mudah sesuai jenis error

• Level Global

  • Error dari broker Kafka/library ditangani secara terpusat melalui opsi wajib xkafka.ErrorHandler
  • Jika handler ini mengembalikan error non-nil, operasi Consumer/Producer akan dihentikan

Kesimpulan

• xkafka mengubah pengalaman penggunaan Apache Kafka yang kompleks menjadi struktur server HTTP yang akrab bagi pengembang Go
• Mengurangi boilerplate yang tidak perlu dan menyediakan lingkungan yang memungkinkan fokus hanya pada logika bisnis
• Jauh lebih ringkas dan intuitif dibandingkan kode confluent-kafka-go yang ada
• Anda bisa langsung mulai dengan melihat dokumentasi dan contoh resmi

[penza1]

Hmm, setahu saya di golang
sarama yang lebih banyak disukai..
Klien Kafka ternyata lebih rumit dari yang saya kira, apalagi saat ada gangguan broker atau pengecualian
apakah bisa mencakup semua kasus..