Poll Loop #

Dalam arsitektur konsumen Apache Kafka, proses membaca data dari broker tidak beroperasi menggunakan model dorong (push model) di mana broker secara aktif membanjiri data ke aplikasi kita. Sebaliknya, Kafka menerapkan model tarik (pull model), di mana aplikasi konsumen kita yang harus secara aktif meminta data ke broker secara terus-menerus. Jantung penggerak dari mekanisme tarik ini adalah Poll Loop (perulangan jajak pendapat). Melalui pemanggilan fungsi tunggal .poll(Duration) di dalam blok perulangan tak terbatas, konsumen kita melakukan koordinasi yang sangat kompleks: dari mulai mendaftarkan diri ke kluster broker, mendeteksi kesehatan anggota kelompok konsumen, memicu proses pembagian partisi ulang (rebalance), melakukan pertukaran detak jantung (heartbeat), hingga mengunduh biner data dari disk broker ke memori cache lokal aplikasi. Memahami cara kerja internal dari siklus poll loop ini sangat krusial agar kita dapat merancang sistem konsumen yang stabil, responsif, dan terbebas dari masalah rebalance liar yang sering melumpuhkan pipeline data produksi.

Cara Kerja Internal Siklus Hidup poll() #

Ketika kita memanggil perintah consumer.poll(Duration.ofMillis(100)) di dalam kode aplikasi kita, di balik layar klien konsumen Kafka (khususnya Java SDK) mengeksekusi serangkaian operasi asinkron yang terstruktur.

Berikut adalah langkah-langkah perjalanan yang terjadi di dalam setiap siklus panggilan .poll():

1. Inisialisasi Group dan Koordinasi Metadata #

Jika konsumen baru saja dinyalakan atau baru bergabung dengan Consumer Group, ia belum memiliki partisi yang dialokasikan untuk dibaca. Panggilan .poll() pertama akan mendeteksi broker Group Coordinator untuk grup tersebut. Konsumen mengirimkan permintaan JoinGroup dan SyncGroup untuk berpartisipasi dalam pembagian partisi. Setelah pembagian disepakati, konsumen memperbarui metadata partisi lokalnya.

2. Membaca Data dari Cache Lokal (Prefetch Buffer) #

Sebelum mengirimkan permintaan baru melalui jaringan, konsumen memeriksa apakah ada data sisa di memori buffer lokal (prefetch buffer).

  • Mekanisme: Pada siklus .poll() sebelumnya, konsumen mungkin telah mengunduh data dalam jumlah besar yang melebihi batas pemrosesan kita. Jika data sisa ini masih ada di cache, .poll() akan langsung mengembalikan data tersebut ke aplikasi kita tanpa melakukan panggilan I/O jaringan ke broker. Ini menghemat bandwidth dan meningkatkan kecepatan pemrosesan lokal.

3. Mengirimkan Fetch Request ke Broker #

Jika cache lokal kosong, konsumen merakit permintaan ambil data (Fetch Request) ke broker yang bertindak sebagai leader dari masing-masing partisi yang dialokasikan.

  • Non-Blocking Network I/O: Permintaan dikirim menggunakan Selector non-blocking Java NIO.
  • Tunggu Data (Timeout Block): Jika broker tidak memiliki pesan baru, panggilan .poll(Duration) akan memblokir thread aplikasi kita secara sinkron sesuai durasi parameter Duration yang kita lewatkan (misalnya menunggu selama 100 milidetik). Jika sebelum 100 ms ada pesan baru masuk ke broker, data langsung dikembalikan. Jika waktu 100 ms habis tanpa ada data baru, .poll() akan mengembalikan koleksi kosong (ConsumerRecords.empty()) agar perulangan aplikasi tetap berputar.

4. Sinkronisasi Heartbeat Thread (Kafka Modern) #

Pada versi Kafka modern (0.10.1+), pengiriman detak jantung (heartbeat) telah dipisahkan ke utas latar belakang terpisah (background heartbeat thread). Namun, pemanggilan .poll() di utas utama tetap bertugas memantau status kesehatan utas latar belakang tersebut dan memastikan tidak ada error fatal yang tidak terdeteksi.

Diagram Mermaid: Alur Kerja poll() Loop Internal #

Bagan alir berikut memvisualisasikan bagaimana aliran logika internal dari klien konsumen Kafka saat mengeksekusi pemanggilan .poll() secara berulang:

flowchart TD
    Start["1. Aplikasi Memanggil consumer.poll(duration)"] --> C1{"Apakah ada data sisa di cache lokal?"}
    
    C1 -- "Ya (Hit)" --> ReturnData["Kembalikan data sisa ke Aplikasi"]
    C1 -- "Tidak (Miss)" --> C2{"Apakah perlu inisialisasi / rebalance?"}
    
    C2 -- "Ya" --> Rebalance["Hubungi Group Coordinator & Jalankan JoinGroup"]
    Rebalance --> Fetch
    
    C2 -- "Tidak" --> Fetch["2. Kirim FetchRequest ke Broker Leader"]
    Fetch --> WaitData{"Apakah data tersedia sebelum timeout?"}
    
    WaitData -- "Ya" --> SaveCache["3. Simpan data di cache lokal"]
    SaveCache --> ReturnData
    
    WaitData -- "Tidak" --> Timeout{"4. Waktu timeout durasi habis"}
    Timeout --> ReturnEmpty["Kembalikan Kumpulan Kosong (Empty Records)"]
    
    ReturnData --> Process["5. Aplikasi Memproses Records (Utas Utama)"]
    ReturnEmpty --> Process
    
    Process --> Loop["6. Kembali ke Awal Loop (Kembali panggil poll)"]
    Loop --> Start
    
    style Start stroke:#e5e7eb
    style Rebalance stroke:#f57c00,stroke-width:2px
    style Fetch stroke:#0288d1,stroke-width:2px
    style ReturnData stroke:#2e7d32,stroke-width:2px
    style ReturnEmpty stroke:#c62828,stroke-width:2px

Pembedahan Arsitektur: Desain Single-Threaded Klien Java #

Salah satu aspek arsitektur terpenting dari konsumen Kafka Java yang wajib dipahami oleh setiap pengembang adalah: KafkaConsumer tidak thread-safe.

Pustaka klien konsumen Kafka dirancang dengan model single-threaded. Artinya, kita tidak boleh menggunakan satu instansi KafkaConsumer secara bersamaan oleh beberapa thread aplikasi yang berbeda untuk melakukan operasi baca (poll()) atau komit (commit()).

Apa yang Terjadi Jika Kita Melanggar Aturan Ini? #

Setiap metode utama di dalam kelas KafkaConsumer memiliki blok pengecekan kepemilikan thread. Jika ia mendeteksi ada thread lain yang mencoba memanggil metode pada instansi yang sama secara paralel, ia akan langsung melemparkan pengecualian ConcurrentModificationException saat runtime.

Dua Pola Desain Multi-Threaded Consumption yang Benar: #

Pola A: Satu Konsumen per Utas (One Consumer Per Thread) #

Kita membuat beberapa thread aplikasi (misal menggunakan Executor Service), dan di dalam masing-masing thread tersebut, kita menginstansiasi satu objek KafkaConsumer independen.

  • Kelebihan: Sangat mudah diimplementasikan, mudah mengelola komit offset karena pemrosesan dan pembacaan berada dalam thread yang sama.
  • Kekurangan: Jumlah koneksi TCP ke broker membengkak (karena tiap thread membuka soket terpisah). Batasan pararelisasi kita terikat mati pada jumlah partisi topik (misal jika topik hanya punya 4 partisi, membuat thread ke-5 akan sia-sia karena ia akan menganggur).

Pola B: Satu Konsumen Pembaca + Utas Pekerja (One Consumer Coordinator + Worker Thread Pool) #

Kita hanya membuat satu thread khusus yang bertugas melakukan poll loop (KafkaConsumer.poll()). Setelah data biner didapatkan, thread pembaca ini langsung melempar tugas pemrosesan bisnis yang berat ke dalam antrean Worker Thread Pool (misal Java ThreadPoolExecutor) secara asinkron. Thread pembaca kemudian langsung kembali memanggil .poll() berikutnya agar tidak dianggap mati oleh broker.

  • Kelebihan: Skalabilitas pemrosesan bisnis sangat fleksibel dan tidak terbatas oleh jumlah partisi topik.
  • Kekurangan: Manajemen komit offset manual menjadi sangat rumit karena utas pembaca harus melacak penyelesaian tugas di seluruh utas pekerja sebelum dapat mengomit offset secara aman tanpa risiko kehilangan data.

Berikut adalah perbandingan kode antara desain multi-thread yang salah (anti-pattern) dengan desain yang benar menggunakan Pola A:

// ANTI-PATTERN: Mengakses satu instansi konsumen dari beberapa thread secara bersamaan
// Memicu ConcurrentModificationException dan merusak status koneksi soket
public class DangerousMultiThreadConsumer {
    private final KafkaConsumer<String, String> consumer;

    public DangerousMultiThreadConsumer(Properties props) {
        this.consumer = new KafkaConsumer<>(props);
        this.consumer.subscribe(Collections.singletonList("orders"));
    }

    public void startConsuming() {
        // ✗ JANGAN: Menjalankan pemrosesan asinkron yang memanggil konsumen secara paralel
        new Thread(() -> {
            while (true) {
                // Thread lain memanggil poll, memicu error ConcurrentModificationException
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                process(records);
            }
        }).start();

        new Thread(() -> {
            while (true) {
                // Thread kedua mencoba mengomit offset menggunakan instansi yang sama
                consumer.commitSync(); 
            }
        }).start();
    }
    private void process(ConsumerRecords<String, String> r) {}
}

// BENAR: Menggunakan pola "One Consumer Per Thread" yang terisolasi secara bersih
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SafeMultiThreadConsumer {
    private final Properties consumerConfigs;
    private final int totalThreads = 3; // Misal topik memiliki 3 partisi
    private ExecutorService executorService;

    public SafeMultiThreadConsumer(Properties props) {
        this.consumerConfigs = props;
    }

    public void start() {
        executorService = Executors.newFixedThreadPool(totalThreads);
        for (int i = 0; i < totalThreads; i++) {
            // ✓ BENAR: Berikan instansi KafkaConsumer baru dan independen untuk setiap thread
            executorService.submit(new ConsumerRunnable(new Properties(consumerConfigs)));
        }
    }

    private static class ConsumerRunnable implements Runnable {
        private final Properties props;

        public ConsumerRunnable(Properties props) {
            this.props = props;
        }

        @Override
        public void run() {
            // Instansi terisolasi di dalam thread ini saja
            try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
                consumer.subscribe(Collections.singletonList("orders"));
                while (!Thread.currentThread().isInterrupted()) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Thread %s memproses order: %s\n", 
                            Thread.currentThread().getName(), record.value());
                    }
                    consumer.commitSync();
                }
            } catch (Exception e) {
                System.err.println("Thread konsumen mati akibat error: " + e.getMessage());
            }
        }
    }
}

Mekanisme Heartbeat dan Pendeteksian Kegagalan #

Bagaimana kluster broker Kafka mengetahui jika salah satu konsumen di dalam kelompok mengalami kecelakaan, crash, mati listrik, atau mengalami kegagalan jaringan? Kafka mengandalkan mekanisme pertukaran Heartbeat (detak jantung).

Di dalam konfigurasi konsumen, terdapat dua parameter terpenting yang mengatur mekanisme deteksi kesehatan ini:

1. Properti session.timeout.ms #

Menentukan batas waktu maksimum (default: 45.000 ms atau 45 detik) bagi broker untuk terus menanti sinyal heartbeat dari konsumen. Jika dalam waktu 45 detik broker tidak menerima satu pun detak jantung dari konsumen tertentu, broker coordinator akan menyimpulkan bahwa konsumen tersebut telah mati. Broker akan segera memicu proses Rebalance untuk memindahkan partisi yang dipegang konsumen mati tersebut ke anggota grup lain yang masih sehat.

2. Properti heartbeat.interval.ms #

Menentukan frekuensi (dalam milidetik) pengiriman sinyal heartbeat oleh utas latar belakang konsumen (background heartbeat thread) ke broker coordinator. Default-nya adalah 3.000 ms (3 detik).

[!IMPORTANT] Aturan Emas Konfigurasi (Rule of Three): Nilai properti heartbeat.interval.ms harus selalu disetel maksimal sepertiga (1/3) dari nilai session.timeout.ms. Hal ini memberikan toleransi bagi konsumen jika 1 atau 2 paket heartbeat hilang di jaringan akibat ketidakstabilan koneksi sesaat, sehingga terhindar dari rebalance palsu. Contoh: Jika session.timeout.ms=45000, maka setel heartbeat.interval.ms=15000.

Menangani Pemrosesan Lambat: max.poll.interval.ms vs max.poll.records #

Bagaimana jika aplikasi konsumen kita tidak mengalami crash fisik (tetap mengirim heartbeat secara rutin di latar belakang), namun thread utama kita mengalami macet total (livelock)? Misalnya thread aplikasi terperangkap dalam perulangan pemrosesan database yang lambat, atau terblokir sinkron menunggu respons API HTTP pihak ketiga yang tidak kunjung kembali.

Untuk mendeteksi skenario pemrosesan lambat ini, Kafka memperkenalkan properti max.poll.interval.ms (default: 300.000 ms atau 5 menit).

Bagaimana Pengecekan Mandek Bekerja? #

  • Thread utama aplikasi mengeksekusi logika bisnis.
  • Thread latar belakang terus mengirimkan heartbeat (detak jantung) secara teratur ke broker (menyatakan “infrastruktur saya masih hidup”).
  • Namun, jika utas utama kita sibuk dan tidak sempat memanggil .poll() berikutnya sebelum batas waktu max.poll.interval.ms (5 menit) terlewati, klien konsumen secara internal akan menyadari bahwa ia mandek (stuck).
  • Klien konsumen akan secara sadar mengirimkan permintaan LeaveGroup ke broker coordinator. Broker akan mencopot partisi konsumen tersebut dan memicu rebalance.

Cara Menyeimbangkan Parameter Pemrosesan #

Jika aplikasi kita sering mengalami rebalance liar akibat terlampauinya batas 5 menit, kita memiliki dua opsi penyelarasan:

  • Opsi 1: Turunkan max.poll.records: Secara default, sekali panggil .poll(), konsumen mengambil maksimal 500 record (diatur oleh max.poll.records). Jika memproses 1 record butuh 1 detik, maka memproses 500 record butuh 500 detik (8,3 menit), melebihi batas interval 5 menit. Dengan menurunkan max.poll.records menjadi 100, total waktu pemrosesan turun menjadi 100 detik (1,6 menit), aman berada di bawah batas timeout.
  • Opsi 2: Naikkan max.poll.interval.ms: Jika kita terpaksa harus memproses batch besar yang memakan waktu lama, naikkan properti max.poll.interval.ms menjadi 10 atau 15 menit.

Implementasi Java: Pola Siklus Poll Loop yang Tangguh #

Untuk menghentikan perulangan poll loop secara bersih saat aplikasi kita dimatikan (graceful shutdown), kita wajib menggunakan metode consumer.wakeup(). Ini adalah satu-satunya metode pada kelas KafkaConsumer yang aman dipanggil dari thread luar secara konkuren.

Berikut adalah contoh implementasi poll loop yang tangguh dengan penanganan shutdown hook yang benar:

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class GracefulShutdownConsumer {
    private static KafkaConsumer<String, String> consumer;
    private static Thread mainThread;

    public static void main(String[] args) {
        mainThread = Thread.currentThread();

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "graceful-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        
        // Optimasi parameter pemrosesan lambat
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100"); // Batasi batch records
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000"); // 5 menit

        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders-topic"));

        // ✓ BENAR: Daftarkan Shutdown Hook pada runtime JVM
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Sinyal shutdown terdeteksi, memicu consumer.wakeup()...");
            
            // Wakeup akan memaksa consumer.poll() yang sedang terblokir melempar WakeupException
            consumer.wakeup(); 
            
            try {
                // Tunggu thread utama selesai menutup koneksi secara bersih
                mainThread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }));

        try {
            while (true) {
                // Blokir nunggu data selama 100ms
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    processBusinessLogic(record);
                }
                
                // Commit offset setelah pemrosesan batch selesai
                consumer.commitSync();
            }
        } catch (WakeupException e) {
            // ✓ BENAR: WakeupException adalah sinyal keluar loop yang sah, abaikan stacktrace-nya
            System.out.println("Consumer berhasil dibangunkan oleh wakeup(). Keluar dari loop...");
        } catch (Exception e) {
            System.err.println("Terjadi error tak terduga pada poll loop: " + e.getMessage());
        } finally {
            // Tutup koneksi soket dan beri tahu Coordinator bahwa kita resmi keluar grup
            System.out.println("Menutup koneksi consumer...");
            consumer.close();
            System.out.println("Consumer ditutup secara graceful.");
        }
    }

    private static void processBusinessLogic(ConsumerRecord<String, String> record) {
        System.out.printf("Memproses transaksi: %s pada offset %d\n", record.value(), record.offset());
    }
}

Ringkasan #

  • Pull Model: Konsumen Kafka beroperasi menggunakan model tarik (pull), mewajibkan aplikasi memanggil .poll() secara berkala untuk mengunduh data.
  • Single Threaded: Klien KafkaConsumer tidak aman secara thread; melanggar aturan ini memicu ConcurrentModificationException instan.
  • Heartbeat Thread: Utas latar belakang terpisah bertugas mengirimkan detak jantung berkala setiap heartbeat.interval.ms untuk menandai keaktifan konsumen.
  • session.timeout.ms: Batas durasi waktu tunggu broker sebelum mencopot kepemilikan partisi dan memicu proses rebalance terhadap konsumen yang tidak merespons.
  • max.poll.interval.ms: Batas waktu antara dua panggilan poll berurutan; jika terlampaui karena pemrosesan lambat, konsumen dianggap mandek dan dikeluarkan.
  • Graceful Shutdown: Gunakan metode thread-safe consumer.wakeup() untuk keluar secara bersih dari perulangan poll loop tanpa merusak status komit offset.

← Sebelumnya: Producer Misconfiguration   Berikutnya: Auto vs Manual Commit →

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