Rebalance Process #

Dalam arsitektur kelompok konsumen (Consumer Group) Apache Kafka, proses pembagian ulang partisi atau Rebalance Process adalah mekanisme krusial yang menjamin ketersediaan tinggi (high availability) dan skalabilitas horisontal secara dinamis. Rebalance otomatis terpicu setiap kali ada konsumen baru yang bergabung, konsumen lama yang sengaja keluar (graceful shutdown), atau ketika konsumen mengalami kegagalan (crash) secara mendadak. Namun, meskipun rebalance adalah fitur keselamatan yang sangat canggih, ia memiliki dampak performa yang signifikan terhadap throughput sistem kita. Memahami siklus hidup internal protokol rebalance, bagaimana mendeteksi pemicunya, serta bagaimana mengimplementasikan listener kustom untuk mengamankan data kita sebelum partisi dipindahkan ke konsumen lain adalah keahlian wajib bagi insinyur sistem terdistribusi.

Mengapa dan Kapan Rebalance Terjadi? #

Proses rebalance bukanlah sesuatu yang terjadi secara acak. Ia adalah tanggapan langsung dari kluster Kafka (khususnya Group Coordinator) terhadap perubahan topologi keanggotaan atau metadata topik.

Berikut adalah lima pemicu utama terjadinya proses rebalance di lingkungan produksi:

Konsumen Baru Bergabung (Scale-Up): Ketika kita menambah pod aplikasi baru untuk membantu mempercepat pemrosesan data, instansi baru tersebut akan mengirimkan permintaan bergabung dengan group.id yang sama, memicu rebalance untuk membagi partisi yang ada agar adil.
Konsumen Keluar Secara Terencana (Graceful Shutdown / Scale-Down): Saat kita melakukan pembaruan aplikasi (rolling upgrade) atau mengurangi kapasitas server, kontainer aplikasi akan ditutup. Sebelum mati, konsumen mengirimkan sinyal LeaveGroup ke Group Coordinator, memicu rebalance agar partisi yang ditinggalkan segera dialihkan ke konsumen yang tersisa.
Absennya Detak Jantung (Unplanned Crash / Network Partition): Jika instansi aplikasi crash akibat mati lampu, kehabisan memori (OutOfMemory), atau terisolasi jaringan, ia akan berhenti mengirimkan heartbeat. Setelah melewati batas session.timeout.ms, Group Coordinator akan menyatakan konsumen tersebut mati dan memulai rebalance.
Livelock (Thread Utama Mandek): Jika utas utama aplikasi kita terlalu lama memproses logika bisnis (misalnya memproses batch query database yang berat) hingga melewati batas waktu max.poll.interval.ms, konsumen akan menganggap dirinya sendiri bermasalah dan mengirimkan instruksi keluar ke broker, memicu rebalance meskipun server secara fisik masih menyala.
Perubahan Metadata Topik: Jika administrator Kafka menambahkan partisi baru ke topik yang sedang kita konsumsi (misalnya melakukan partition expansion dari 10 menjadi 20 partisi), rebalance akan terjadi agar konsumen dapat mendistribusikan partisi baru tersebut.

Siklus Hidup Protokol Rebalance Internal #

Untuk memahami bagaimana data terhenti selama rebalance, kita harus melihat perbedaan antara dua protokol rebalance utama: Eager Rebalance (protokol lama) dan Cooperative Rebalance (protokol baru).

1. Eager Rebalance (Protokol Klasik) #

Protokol ini menerapkan prinsip “lepas semua, cari kembali”. Berikut adalah fase demi fase perjalanannya:

sequenceDiagram
    participant C1 as Consumer 1 (Leader)
    participant C2 as Consumer 2
    participant GC as Group Coordinator (Broker)

    Note over C1,C2: Fase 1: Deteksi Pemicu
    C2->>GC: LeaveGroup Request (C2 dimatikan)
    GC->>C1: Deteksi perubahan metadata saat Heartbeat / Poll

    Note over C1: Fase 2: Pelepasan Partisi (Revocation)
    C1->>C1: Hentikan pemrosesan data, lepaskan seluruh partisi (Stop-The-World)
    
    Note over C1: Fase 3: Mengirim JoinGroup Request
    C1->>GC: JoinGroup Request
    GC-->>C1: JoinGroup Response (C1 ditunjuk sebagai Group Leader)

    Note over C1: Fase 4: Kalkulasi Pembagian Partisi
    C1->>C1: Hitung ulang alokasi partisi untuk seluruh anggota yang tersisa

    Note over C1: Fase 5: Sinkronisasi (SyncGroup)
    C1->>GC: SyncGroup Request (Mengirimkan hasil kalkulasi alokasi baru)
    GC-->>C1: SyncGroup Response (Menerima alokasi baru)

Selama proses dari Fase 2 hingga Fase 5 selesai, seluruh konsumen tidak dapat memproses data baru. Fase pemrosesan berhenti secara total (Stop-The-World). Jika kluster memiliki ratusan partisi dan puluhan konsumen, proses ini bisa memakan waktu hingga puluhan detik, memicu penumpukan antrean (lag spike).

2. Cooperative Rebalance (Incremental) #

Protokol modern ini (diperkenalkan sejak Kafka 2.4) beroperasi secara kooperatif dan bertahap. Alih-alih melucuti seluruh partisi dari seluruh konsumen secara serentak, protokol ini membagi perpindahan menjadi dua siklus rebalance kecil tanpa mengganggu pemrosesan:

Siklus Pertama: Ketika ada perubahan keanggotaan (misalnya konsumen baru C3 bergabung), Group Coordinator dihubungi. Klien menghitung alokasi baru dan menyadari bahwa partisi P2 yang sebelumnya dipegang oleh C2 harus dipindahkan ke C3.
Pencabutan Parsial: Klien konsumen C2 melepaskan partisi P2 saja. Pada fase ini, konsumen C1 yang memegang P0 dan P1 tidak dipengaruhi sama sekali dan tetap terus memproses data tanpa henti.
Siklus Kedua: Setelah P2 terbebas, rebalance kedua yang sangat cepat dieksekusi hanya untuk menugaskan P2 ke konsumen baru C3. Siklus ini berlangsung sangat cepat dalam hitungan milidetik karena tidak ada pelepasan partisi global.

State Machine Consumer Group di Sisi Broker Coordinator #

Di dalam broker yang bertindak sebagai Group Coordinator, terdapat mesin status (state machine) internal yang mengelola transisi keanggotaan grup. Memahami transisi ini akan mempermudah kita saat membaca file log broker Kafka (server.log) ketika melakukan debugging.

Berikut adalah lima status (states) keanggotaan grup di dalam Kafka:

Empty: Grup tidak memiliki anggota aktif, namun metadata offset komit terakhir untuk grup tersebut masih tersimpan di dalam topik internal __consumer_offsets. Keadaan ini terjadi jika seluruh pod aplikasi dimatikan.
PreparingRebalance: Group Coordinator telah menerima pemicu (seperti hilangnya heartbeat atau request bergabung baru) dan bersiap untuk mendistribusikan ulang partisi. Broker sedang menunggu seluruh konsumen aktif mengirimkan request JoinGroup sebelum batas tenggat waktu.
CompletingRebalance: Seluruh anggota aktif telah mengirimkan request JoinGroup. Coordinator menunjuk Group Leader dan sedang menunggu sang leader mengembalikan hasil kalkulasi pembagian partisi melalui request SyncGroup.
Stable: Transaksi pembagian selesai. Seluruh konsumen aktif telah menerima alokasi partisi masing-masing dan sedang aktif membaca data serta mengirimkan heartbeat berkala. Ini adalah keadaan ideal sistem produksi.
Dead: Grup dianggap mati secara permanen karena tidak ada aktivitas dalam waktu lama, atau grup sedang dihapus secara manual oleh administrator sistem melalui perintah CLI.

Bahaya Rebalance Storm (Badai Pembagian Ulang) #

Salah satu fenomena paling berbahaya di lingkungan produksi berskala besar adalah Rebalance Storm (Badai Pembagian Ulang). Kondisi ini terjadi ketika grup konsumen masuk ke dalam siklus rebalance tiada henti yang melumpuhkan sistem.

Pemicu Badai: Jeda Garbage Collection (GC Pauses) #

Pada aplikasi berbasis Java Virtual Machine (JVM) seperti Java, Scala, atau Kotlin, proses pengelolaan memori otomatis menggunakan Garbage Collector. Jika aplikasi kita mengalami penumpukan objek sampah yang besar di memori heap, JVM akan memicu siklus pembersihan menyeluruh yang dikenal sebagai Stop-The-World Garbage Collection (Major GC).

Selama proses Major GC berlangsung, seluruh utas JVM (termasuk utas heartbeat latar belakang milik klien Kafka) akan dibekukan total oleh CPU.
Jika jeda GC ini berlangsung selama 50 detik, sedangkan konfigurasi session.timeout.ms kita setel sebesar 45 detik, broker coordinator akan menyimpulkan bahwa konsumen kita telah mati karena tidak menerima detak jantung dalam 45 detik.
Broker memicu proses rebalance untuk memindahkan partisi ke pod konsumen lain.
Ketika jeda GC selesai, konsumen pertama menyadari bahwa ia telah dikeluarkan dari grup secara paksa. Ia segera mengirimkan permintaan bergabung kembali (JoinGroup).
Permintaan gabung kembali ini memicu rebalance baru lagi.
Selama rebalance baru berjalan, beban kerja berpindah ke pod lain yang dapat memicu lonjakan memori di pod tersebut, memicu Major GC di pod kedua, dan siklus bencana ini berulang terus-menerus menyebar ke seluruh pod konsumen.

Rekomendasi Pencegahan Rebalance Storm #

Untuk mengamankan aplikasi kita dari ancaman Rebalance Storm, kita disarankan untuk menerapkan langkah-langkah mitigasi berikut:

Gunakan G1GC atau ZGC: Konfigurasikan JVM untuk menggunakan Garbage Collector modern yang dirancang untuk meminimalkan jeda waktu pembekuan utas (misalnya menggunakan opsi -XX:+UseG1GC atau -XX:+UseZGC).
Tingkatkan Session Timeout: Jangan menyetel session.timeout.ms terlalu dekat dengan interval heartbeat. Di lingkungan produksi yang padat, angka 45000 (45 detik) hingga 60000 (60 detik) adalah pilihan yang aman.
Optimalkan Heap Memory: Lakukan profiling memori secara rutin untuk menghindari kebocoran memori (memory leaks) yang memicu GC beruntun.

Rebalance Listener: Menyelamatkan State Sebelum Rebalance #

Salah satu kesalahan paling fatal saat mengelola offset secara manual atau menyimpan state lokal di memori aplikasi adalah tidak merespons peristiwa rebalance dengan benar. Ketika Kafka mencabut hak akses partisi dari aplikasi kita untuk diserahkan ke konsumen lain, kita harus memastikan bahwa:

Seluruh pesan yang sedang diproses selesai dieksekusi secara tertib.
Offset komit terakhir sukses dikirim ke broker sebelum partisi tersebut dipegang oleh orang lain (mencegah duplikasi).
Cache memori lokal yang berhubungan dengan partisi tersebut dibersihkan dengan aman.

Untuk menangani kebutuhan ini, Kafka SDK menyediakan interface ConsumerRebalanceListener. Listener ini memiliki dua fungsi callback utama:

1. `onPartitionsRevoked(Collection<TopicPartition> partitions)` #

Kapan dipanggil: Tepat sebelum partisi secara resmi dicabut dari konsumen saat ini.
Tujuan: Di sinilah tempat kita wajib melakukan commit offset secara sinkron (commitSync()) untuk data terakhir yang selesai kita proses. Karena partisi ini akan segera dikonsumsi oleh node lain, mengomit offset di sini adalah benteng pertahanan terakhir kita dari bencana duplikasi data.

2. `onPartitionsAssigned(Collection<TopicPartition> partitions)` #

Kapan dipanggil: Sesaat setelah partisi baru dialokasikan ke konsumen saat ini (sebelum pembacaan data via .poll() dimulai kembali).
Tujuan: Di sini kita bisa melakukan inisialisasi state lokal, mencari posisi offset kustom dari database eksternal menggunakan consumer.seek(), atau mencatat log metrik monitoring.

Kode Penerapan: Mengamankan Offset via Rebalance Listener #

Mari kita lihat perbedaan implementasi kode antara penulisan konsumen manual tanpa penanganan rebalance (yang memicu kebocoran data) dengan penerapan Listener kustom yang kokoh.

Java SDK Anti-Pattern: Mengabaikan Peristiwa Rebalance pada Manual Commit #

Jika kita mematikan auto-commit tetapi tidak mendaftarkan Rebalance Listener, setiap kali terjadi rebalance (misalnya saat kita men-deploy ulang aplikasi), pesan yang sedang diproses di memori akan dikonsumsi ulang oleh konsumen baru karena offset terakhir belum sempat dikomit ke broker.

// ANTI-PATTERN: Mengonsumsi data dengan manual commit tanpa menangani rebalance
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-worker");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Manual commit aktif
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// ✗ JANGAN: Hanya berlangganan topik tanpa melampirkan Rebalance Listener.
// Jika rebalance terjadi di tengah-tengah pemrosesan batch, offset komit terakhir
// akan hangus dan memicu pemrosesan ganda masif di pod yang baru.
consumer.subscribe(Collections.singletonList("order-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processOrder(record.value());
        }
        // Commit dilakukan di akhir perulangan batch
        consumer.commitSync();
    }
} finally {
    consumer.close();
}

Java SDK Solusi: Menerapkan ConsumerRebalanceListener secara Tertib #

Dengan mengimplementasikan ConsumerRebalanceListener, kita menangkap momen sesaat sebelum partisi dipindahkan untuk memaksa komit offset terakhir secara sinkron, serta melacak kembali posisi baca partisi baru secara akurat.

// BENAR: Menggunakan ConsumerRebalanceListener untuk sinkronisasi offset yang aman
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-worker");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Map lokal untuk melacak offset terakhir yang belum dikomit secara memori
final Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("Proses Rebalance Terdeteksi! Partisi akan dicabut: {}", partitions);
        
        // ✓ BENAR: Segera komit offset terakhir yang sedang diproses di memori secara sinkron.
        // Ini menjamin konsumen baru yang mengambil alih partisi ini akan melanjutkan
        // dari offset yang benar, menghindari pemrosesan data ganda.
        try {
            consumer.commitSync(currentOffsets);
        } catch (Exception e) {
            log.error("Gagal melakukan commit offset darurat saat rebalance", e);
        }
        
        // Bersihkan tracking offsets untuk partisi yang dicabut
        currentOffsets.keySet().removeAll(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("Partisi baru berhasil ditugaskan ke konsumen ini: {}", partitions);
        
        // Di sini kita bisa melakukan inisialisasi resources atau seek() jika diperlukan.
        // Contoh: Memastikan pointer pembacaan dimulai tepat setelah offset komit di broker.
        for (TopicPartition partition : partitions) {
            OffsetAndMetadata committed = consumer.committed(partition);
            if (committed != null) {
                log.info("Memulai partisi {} dari committed offset {}", partition, committed.offset());
                consumer.seek(partition, committed.offset());
            }
        }
    }
};

// ✓ Daftarkan subscribe dengan menyertakan listener kustom
consumer.subscribe(Collections.singletonList("order-events"), rebalanceListener);

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processOrder(record.value());
            
            // Lacak offset record berikutnya yang akan dikonsumsi (+1)
            currentOffsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }
        
        if (!records.isEmpty()) {
            consumer.commitAsync(currentOffsets, null);
        }
    }
} catch (Exception e) {
    log.error("Terjadi error fatal di loop konsumen", e);
} finally {
    try {
        // Commit akhir secara darurat sebelum penutupan koneksi
        consumer.commitSync(currentOffsets);
    } finally {
        consumer.close();
    }
}

Ringkasan #

Rebalance Trigger — Peristiwa pembagian ulang partisi yang dipicu oleh penambahan konsumen baru, matinya konsumen lama secara mendadak, atau perubahan metadata partisi topik.

Stop-The-World Effect — Pada protokol Eager Rebalance klasik, seluruh pemrosesan data dihentikan sementara secara global saat seluruh partisi dicabut dan dialokasikan ulang dari awal.

Cooperative Rebalance — Protokol modern (incremental) yang meminimalkan interupsi pemrosesan dengan hanya mencabut partisi yang berpindah kepemilikan tanpa mereset status konsumen stabil lainnya.

ConsumerRebalanceListener — Interface interseptor wajib yang memungkinkan aplikasi menangkap peristiwa pencabutan dan alokasi partisi secara real-time.

onPartitionsRevoked — Callback krusial untuk segera melakukan komit offset secara sinkron (commitSync()) sesaat sebelum kepemilikan partisi berpindah tangan guna mencegah risiko duplikasi data.

onPartitionsAssigned — Callback pasca-rebalance yang digunakan untuk inisialisasi state, membersihkan cache, atau melompati posisi baca offset (seek) dari penyimpanan database eksternal.

Rebalance Storm Mitigations — Jeda GC panjang pada JVM dapat memicu rebalance storm liar, yang dapat ditanggulangi dengan menaikkan session timeout atau beralih ke engine ZGC/G1GC.

Graceful Shutdown — Selalu memanggil fungsi .close() di blok finally aplikasi agar coordinator segera memicu rebalance tanpa perlu menunggu berakhirnya session timeout.

← Sebelumnya: Partition Assignment Strategy Berikutnya: At-Least-Once vs At-Most-Once →