Consumer #

Jika produsen bertugas menulis event ke dalam kluster Apache Kafka, maka Consumer (konsumen) adalah aplikasi klien yang berlangganan (subscribe) ke satu atau beberapa topik dan membaca data tersebut untuk diproses. Berbeda dengan model dorong (push-based) pada broker tradisional, Kafka mengadopsi model tarik (pull-based) yang unik. Desain ini memberikan kendali penuh kepada konsumen atas kecepatan pemrosesan mereka sendiri. Memahami cara kerja poll loop konsumen, mekanisme koordinasi kelompok (Consumer Group), dan pengelolaan komitmen posisi baca (offset commit) sangatlah penting untuk membangun pipa data terdistribusi yang aman, efisien, dan bebas dari duplikasi data atau kehilangan pesan.

Konsep Dasar: Model Pull vs Push #

Dalam arsitektur pengiriman pesan terdistribusi, ada dua metode utama yang digunakan untuk mengirimkan data dari server broker ke aplikasi penerima: Push (didorong oleh broker) dan Pull (ditarik oleh konsumen).

  • Model Push (Tradisional): Broker pesan seperti RabbitMQ secara aktif memantau antrean dan langsung mengirimkan (push) data ke aplikasi konsumen begitu ada pesan baru masuk. Masalah terbesar dari model ini adalah ancaman Overload (kelebihan beban) di sisi konsumen. Jika produsen mendadak mengirimkan 100.000 pesan per detik, sedangkan konsumen hanya mampu memproses 1.000 pesan per detik, broker akan membanjiri konsumen dengan data hingga memori konsumen habis dan aplikasi mati mendadak (Out Of Memory). Untuk mencegah ini, sistem harus menerapkan mekanisme negosiasi kecepatan (backpressure) yang rumit.
  • Model Pull (Kafka): Broker Kafka bersikap pasif. Aplikasi konsumen lah yang aktif mengirimkan permintaan penarikan data (pull request) ke broker ketika mereka memiliki kapasitas komputasi yang siap. Jika konsumen sedang sibuk memproses transaksi berat, konsumen cukup menunda pengiriman permintaan berikutnya. Broker Kafka tidak akan pernah memaksa mengirimkan data. Model ini secara alami memecahkan masalah backpressure karena konsumen sendiri yang menentukan kecepatan alur kerjanya sesuai kapasitas memori dan CPU lokal mereka.

Siklus Kerja Consumer: Poll Loop #

Di dalam kode aplikasi konsumen Kafka, jantung dari seluruh aktivitas membaca data adalah sebuah siklus perulangan tanpa henti yang disebut Poll Loop. SDK konsumen dirancang untuk dijalankan pada satu thread tunggal (single-thread execution model) yang melakukan panggilan berulang ke metode poll() di dalam blok while(true).

Mari kita visualisasikan siklus hidup poll loop di dalam thread konsumen Kafka:

flowchart TD
    subgraph Loop ["Siklus Poll Loop Konsumen"]
        direction TB
        CallPoll["1. Panggil poll(timeout)"] --> FetchData["2. Tarik Event dari Broker <br/> (Ambil batch pesan)"]
        FetchData --> ProcessData["3. Proses Bisnis Utama <br/> (Simpan ke DB, hitung data, dll)"]
        ProcessData --> SendHeartbeat["4. Kirim Heartbeat Latar Belakang <br/> (Beri tahu broker jika konsumen hidup)"]
        SendHeartbeat --> CommitOffset["5. Commit Offset Bacaan <br/> (Catat kemajuan ke __consumer_offsets)"]
        CommitOffset --> CallPoll
    end

Siklus di atas bekerja secara terus-menerus. Panggilan poll(timeout) memiliki dua peran utama:

  1. Mengambil Data: Panggilan ini mengambil batch pesan baru dari partisi-partisi yang dialokasikan ke konsumen tersebut. Jika tidak ada pesan baru di broker, thread akan menunggu selama batas waktu timeout (misalnya 100 ms) sebelum melanjutkan ke perulangan berikutnya agar tidak memboroskan CPU.
  2. Menjaga Keanggotaan Kelompok: Di balik layar, panggilan poll() secara berkala mengirimkan sinyal detak jantung (heartbeat) ke server broker coordinator. Ini adalah mekanisme penting untuk memberi tahu kluster bahwa aplikasi konsumen kita masih berjalan dengan sehat dan tidak mati.

Kelompok Konsumen (Consumer Group) & Skalabilitas Baca #

Untuk menangani aliran data bervolume besar secara real-time, satu aplikasi konsumen tunggal sering kali tidak cukup. Kita perlu menjalankan beberapa instance konsumen sekaligus secara paralel untuk membagi beban kerja. Kafka memfasilitasi hal ini secara bawaan melalui konsep Consumer Group (Kelompok Konsumen).

Semua instance konsumen yang dikonfigurasi dengan properti group.id yang sama akan otomatis bergabung ke dalam kelompok yang sama. Kafka kemudian akan membagi seluruh partisi aktif yang ada di dalam topik tersebut kepada para anggota kelompok secara adil dan merata.

Mari kita pelajari tiga skenario alokasi partisi fisik berdasarkan jumlah anggota di dalam satu kelompok konsumen:

flowchart TD
    subgraph Skenario1 ["Skenario A: Konsumen < Partisi (Ideal)"]
        direction LR
        P0_A[("Partisi 0")] ---> C1_A["Konsumen 1"]
        P1_A[("Partisi 1")] ---> C1_A
        P2_A[("Partisi 2")] ---> C2_A["Konsumen 2"]
    end

    subgraph Skenario2 ["Skenario B: Konsumen = Partisi (Maksimal)"]
        direction LR
        P0_B[("Partisi 0")] ---> C1_B["Konsumen 1"]
        P1_B[("Partisi 1")] ---> C2_B["Konsumen 2"]
        P2_B[("Partisi 2")] ---> C3_B["Konsumen 3"]
    end

    subgraph Skenario3 ["Skenario C: Konsumen > Partisi (Mubazir)"]
        direction LR
        P0_C[("Partisi 0")] ---> C1_C["Konsumen 1"]
        P1_C[("Partisi 1")] ---> C2_C["Konsumen 2"]
        P2_C[("Partisi 2")] ---> C3_C["Konsumen 3"]
        P3_Idle["(Tidak ada partisi)"] -.-> C4_C["Konsumen 4 <br/> (Menganggur / Idle)"]
    end

Aturan Alokasi Partisi: #

  • Satu Partisi, Satu Konsumen: Di dalam satu Consumer Group yang sama, setiap partisi hanya boleh dialokasikan ke maksimal satu konsumen. Ini mutlak diperlukan untuk mencegah perebutan data dan menjamin urutan pesan per partisi tetap terjaga.
  • Instance Berlebih Menganggur: Jika jumlah instance konsumen kita melebihi total jumlah partisi topik (seperti pada Skenario C di atas), instance tambahan tersebut akan menganggur (idle) tanpa menerima data sama sekali. Mereka hanya bertindak sebagai cadangan aktif (hot standby) yang siap mengambil alih partisi secara instan jika salah satu konsumen aktif mengalami crash.

Mekanisme Rebalance (Rebalancing) #

Proses penataan ulang kepemilikan partisi di antara para konsumen di dalam kelompok disebut dengan istilah Rebalance. Proses ini dikoordinasikan secara otomatis oleh salah satu broker Kafka yang ditunjuk sebagai Group Coordinator.

Rebalance akan dipicu secara otomatis oleh kluster ketika terjadi perubahan struktur keanggotaan kelompok:

  • Ada instance konsumen baru yang bergabung ke dalam kelompok.
  • Ada instance konsumen aktif yang keluar dari kelompok secara sengaja (aplikasi dimatikan secara bersih).
  • Ada instance konsumen yang dianggap mati oleh broker karena gagal mengirim detak jantung dalam batas waktu tertentu.
  • Terjadi perubahan pada topik itu sendiri (misalnya jumlah partisi topik bertambah).

Parameter Kritis yang Mengontrol Rebalance #

Untuk menghindari fenomena rebalance liar yang tidak perlu (yang dapat menghentikan pemrosesan data sementara waktu/stop-the-world), kita harus menyetel parameter konfigurasi berikut dengan sangat hati-hati:

  • session.timeout.ms: Batas waktu maksimum broker menunggu detak jantung dari konsumen sebelum menganggap konsumen tersebut mati (default 45.000 ms atau 45 detik). Jika koneksi jaringan terputus sementara melewati batas ini, broker akan menendang konsumen tersebut dari grup dan memicu rebalance.
  • heartbeat.interval.ms: Seberapa sering konsumen mengirim detak jantung ke broker (default 3.000 ms atau 3 detik). Aturan praktisnya adalah mengatur nilai ini sebesar sepertiga dari session.timeout.ms.
  • max.poll.interval.ms: Batas waktu maksimum toleransi jarak antar-panggilan poll() di sisi konsumen (default 300.000 ms atau 5 menit). Jika aplikasi konsumen kita membutuhkan waktu memproses satu batch data melebihi 5 menit (misalnya karena terhambat query database yang lambat), konsumen akan sengaja keluar dari grup karena dianggap macet (stuck), memicu rebalance yang berulang-ulang secara terus-menerus.

Protokol Rebalance: Eager vs Cooperative (Incremental) #

Secara historis, Kafka menggunakan protokol Eager Rebalance (yang bersifat destruktif). Di bawah protokol ini, ketika rebalance terjadi, seluruh konsumen dalam kelompok wajib menghentikan aktivitas membaca data (stop-the-world), melepaskan semua partisi yang sedang dikelolanya, dan menunggu broker Coordinator menetapkan alokasi partisi yang baru dari awal. Hal ini menurunkan performa sistem secara drastis saat rebalance terjadi pada kluster skala besar.

Sejak Kafka 2.4, diperkenalkan protokol Cooperative Rebalancing (atau Incremental Cooperative Rebalancing). Di bawah protokol modern ini, hanya partisi-partisi yang memang perlu dipindahkan ke konsumen lain yang akan dilepaskan, sedangkan partisi yang tetap dialokasikan pada konsumen lama akan terus berjalan memproses data secara asinkron tanpa henti. Ini sangat menghemat waktu tunggu rebalancing dan meminimalkan gangguan kinerja sistem produksi.

Arsitektur Pemisahan Thread: Heartbeat vs Poll #

Banyak pengembang bingung mengapa parameter session.timeout.ms dan max.poll.interval.ms dipisahkan. Sejak Kafka 0.10.1, SDK konsumen menggunakan arsitektur dua thread internal:

  1. Poll Thread: Thread utama yang mengeksekusi kode logika bisnis kita dan memanggil poll(). Jika thread ini macet (misalnya karena deadlock atau I/O database lambat), ia akan melebihi batas max.poll.interval.ms.
  2. Heartbeat Thread: Thread latar belakang khusus yang berjalan otomatis di belakang layar untuk mengirim detak jantung secara periodik ke broker selama poll thread masih aktif (belum dianggap mati di tingkat OS). Jika seluruh proses aplikasi kita mengalami hard crash (misalnya JVM crash), heartbeat thread akan mati, dan broker mendeteksinya setelah melewati session.timeout.ms.

Pemisahan ini memungkinkan kita menyetel timeout detak jantung yang sensitif terhadap kegagalan server fisik (misal 10-30 detik) tanpa takut terganggu oleh waktu pemrosesan batch data lokal yang panjang (misal 5 menit).

Manajemen Offset & Komitmen (Offset Commit) #

Ketika aplikasi konsumen membaca data dari partisi, konsumen harus mencatat kemajuan bacanya secara berkala agar jika aplikasi mengalami crash dan dinyalakan kembali, ia tidak perlu membaca ulang seluruh data dari awal. Pencatatan kemajuan ini disebut sebagai Offset Commit (komitmen offset).

Kafka menyimpan data offset komitmen ini di dalam topik internal kluster khusus bernama __consumer_offsets. Ada dua cara utama untuk mengelola komitmen offset ini:

1. Komitmen Otomatis (Auto Commit) #

Diaktifkan dengan menyetel enable.auto.commit = true. Konsumen akan secara otomatis mengirimkan komitmen offset terakhir yang ia terima ke broker setiap interval waktu tertentu (diatur oleh auto.commit.interval.ms, default 5 detik).

  • Kelebihan: Sangat mudah digunakan karena tidak memerlukan baris kode tambahan.
  • Kekurangan: Berisiko tinggi memicu kehilangan data (data loss) atau duplikasi data. Jika aplikasi kita memanggil poll(), menerima data, lalu broker otomatis mengomit offset tersebut 5 detik kemudian meskipun aplikasi kita mengalami crash di tengah jalan sebelum sukses menyelesaikan kalkulasi bisnis data, data tersebut dianggap sukses diproses oleh Kafka dan tidak akan dikirimkan kembali saat aplikasi hidup lagi.

2. Komitmen Manual (Manual Commit) #

Diaktifkan dengan menyetel enable.auto.commit = false. Kita sebagai pengembang memegang kendali penuh kapan harus memberi tahu Kafka bahwa data telah sukses diproses melalui kode aplikasi. SDK Kafka menyediakan dua metode:

  • commitSync(): Memblokir thread eksekusi hingga broker selesai merespon status penyimpanan offset. Ini sangat aman namun menurunkan throughput karena memicu waktu tunggu sinkron.
  • commitAsync(): Mengirimkan komitmen offset secara asinkron tanpa memblokir thread. Ini sangat cepat, namun memiliki risiko jika terjadi kegagalan jaringan, offset yang lebih lama dapat menimpa offset yang lebih baru jika urutan respon terbalik.

Mendeteksi dan Mengatasi Consumer Lag #

Dalam memantau kesehatan pipa data Apache Kafka, metrik terpenting yang wajib kita awasi secara real-time adalah Consumer Lag (keterlambatan konsumen).

Consumer Lag didefinisikan sebagai selisih (jarak) antara offset pesan terbaru yang sukses ditulis oleh produsen di disk broker (Log End Offset) dengan offset pesan terakhir yang sukses diproses dan dikomit oleh kelompok konsumen kita (Current Offset):

$$\text{Consumer Lag} = \text{Log End Offset} - \text{Current Offset}$$

Jika nilai lag terus meningkat secara konstan dari waktu ke waktu, ini merupakan alarm merah yang menandakan bahwa konsumen kita tidak mampu mengimbangi kecepatan pengiriman data produsen.

Cara Mengatasi Consumer Lag yang Tinggi: #

  1. Tingkatkan Paralelisme (Scale Up): Tambahkan jumlah instance konsumen baru di dalam kelompok konsumen tersebut. Pastikan jumlah partisi topik cukup untuk menampung konsumen baru (jika partisi hanya 3 dan konsumen sudah 3, penambahan konsumen ke-4 akan sia-sia; kita harus menaikkan jumlah partisi topik terlebih dahulu).
  2. Batch Tuning: Tingkatkan nilai max.poll.records (jumlah pesan maksimum yang diambil dalam satu kali panggilan poll) agar konsumen dapat memproses data dalam volume batch yang lebih besar sekaligus.
  3. Offload I/O Blocking (Worker Thread Pattern): Jika pemrosesan data memerlukan panggilan I/O yang lambat (seperti memanggil REST API eksternal), pisahkan thread penarik data dengan thread pemroses data menggunakan antrean memori internal (in-memory queue) di tingkat aplikasi.

Kesalahan Umum (Anti-pattern) dalam Penggunaan Consumer #

Berikut adalah kesalahan implementasi di tingkat aplikasi konsumen yang sering kali berakibat fatal bagi kestabilan kluster:

1. Melakukan Proses Bisnis Berat Langsung di Thread Poll Utama #

Pengembang menulis logika yang memakan waktu lama, seperti memproses konversi video, melakukan pengunduhan file besar, atau melakukan query laporan SQL kompleks langsung di dalam loop while(true) penarik data.

Konsekuensi Kegagalan: Thread terblokir terlalu lama dan gagal memanggil metode poll() berikutnya sebelum batas waktu max.poll.interval.ms (5 menit) habis. Broker Coordinator akan menganggap konsumen tersebut mati, mengeluarkannya dari kelompok, dan memicu proses rebalance yang menghentikan aktivitas baca konsumen lain. Saat konsumen yang terblokir tersebut akhirnya selesai memproses dan mencoba memanggil poll(), ia akan terkejut karena posisinya sudah digantikan, memicu error CommitFailedException dan siklus rebalance tanpa akhir.

# ANTI-PATTERN: Memproses logika berat yang memblokir thread poll utama
# Tindakan ini memicu kick-out dari Consumer Group karena melebihi max.poll.interval.ms.
def jalankan_consumer_salah(consumer):
    while True:
        records = consumer.poll(timeout_ms=100)
        for record in records:
            # MEMPROSES LOGIKA BERAT YANG MEMAKAN WAKTU 10 MENIT SINKRON
            # JANGAN LAKUKAN INI! Thread poll utama akan terblokir.
            proses_analisis_laporan_berat(record.value)
            consumer.commit()

# Solusi yang BENAR: Pola Worker Thread Pool (Asinkron)
# Thread poll utama hanya bertugas cepat menarik data dan langsung menyerahkannya ke executor thread pool.
from concurrent.futures import ThreadPoolExecutor
import queue

worker_pool = ThreadPoolExecutor(max_workers=10)

def jalankan_consumer_benar(consumer):
    while True:
        # Panggilan poll() selalu berjalan cepat dan lancar secara non-blocking
        records = consumer.poll(timeout_ms=100)
        for record in records:
            # Menyerahkan tugas berat ke thread pool asinkron
            worker_pool.submit(proses_analisis_laporan_berat, record.value)
            
            # Offset dikomit secara manual setelah koordinasi status selesai dilakukan
            consumer.commit_async()

Ringkasan #

  • Model Pull — Consumer Kafka aktif menarik data (pull) dari broker sesuai dengan kesiapan kapasitas lokal mereka sendiri, mencegah terjadinya kelebihan beban (backpressure) secara alami.
  • Siklus Poll — Konsumen berjalan pada satu thread tunggal dalam siklus poll loop yang terus memanggil poll() untuk mengambil data baru sekaligus mengirimkan detak jantung (heartbeat) koordinasi kelompok.
  • Consumer Group — Konsep kelompok konsumen memungkinkan partisi-partisi dari satu topik dibagi secara merata kepada beberapa instance konsumen untuk mencapai paralelisme baca yang tinggi.
  • Mekanisme Rebalance — Proses pembagian ulang partisi yang dipicu ketika ada konsumen baru bergabung, konsumen keluar, atau konsumen dianggap mati karena gagal mengirim detak jantung melewati batas session.timeout.ms.
  • Manajemen Offset — Kemajuan baca dicatat di topik internal __consumer_offsets. Hindari penggunaan Auto Commit untuk data penting; selalu gunakan Manual Commit pasca-proses bisnis sukses dilakukan demi keamanan data.
  • Deteksi Lag — Pantau kesehatan konsumen dengan mengawasi metrik Consumer Lag. Lag yang terus naik menuntut tindakan penskalaan instance konsumen atau pengalihan proses berat ke Worker Thread Pool terpisah.

← Sebelumnya: Producer   Berikutnya: Broker →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact