Consumer Groups

Ketika traffic pesanan TokoKita melewati kapasitas satu proses consumer, solusinya bukan mempercepat consumer tersebut, melainkan menambah consumer baru. Consumer group adalah mekanisme Kafka untuk membagi partisi topic ke sejumlah consumer yang bekerja paralel: setiap partisi hanya dimiliki satu anggota, sehingga tidak ada pesan yang diproses dua kali oleh group yang sama. Keanggotaan group bersifat dinamis — consumer bergabung, keluar, dan gagal — dan setiap perubahan memicu rebalance yang menghentikan seluruh group sejenak. Chapter ini membedah protokol rebalance, lalu mensimulasikan rolling restart dengan dua gaya keanggotaan: dinamis (bawaan) dan statis.

Satu Group, Banyak Consumer #

Group didefinisikan lewat satu konfigurasi: group.id. Semua consumer dengan group yang sama dihitung satu tim kerja; broker menugaskan setiap partisi topic ke tepat satu anggota group sehingga beban terbagi tanpa koordinasi manual di sisi aplikasi. Dua group berbeda yang membaca topic yang sama masing-masing menerima salinan penuh — pola publish/subscribe — sementara satu group bekerja seperti antrian kerja terbagi. Keanggotaan dilacak oleh satu broker sebagai group coordinator, terpilih dari internal topic __consumer_offsets; coordinator menerima heartbeat tiap anggota dan memulai rebalance begitu keanggotaan berubah.

Untuk topic enam partisi yang kita pakai sepanjang chapter ini, pembagian bebannya berbentuk seperti ini:

Anggota yang menganggur tetap ikut protokol group namun tidak menerima partisi; menambah consumer di atas jumlah partisi hanya menambah beban koordinasi tanpa menambah throughput. Konfigurasi lengkap sisi consumer terdokumentasi di dokumentasi Apache Kafka .

Protokol Rebalance: Stop-the-World #

Rebalance adalah kesepakatan ulang penugasan partisi. Protokol klasiknya dua tahap: seluruh anggota mengirim JoinGroup ke coordinator, coordinator menunjuk satu anggota sebagai group leader, lalu leader menghitung rencana penugasan dan mengedarkannya lewat SyncGroup.

Selama dua tahap ini, tidak ada anggota yang boleh mem-poll: seluruh group berhenti sejenak (stop-the-world). Rebalance dipicu empat hal: consumer baru bergabung, consumer keluar dengan sopan lewat LeaveGroup, consumer gagal hingga session.timeout.ms (bawaan 45 detik pada client Java/C) habis tanpa heartbeat, atau daftar langganan berubah.

Simulasi: Rolling Restart pada Enam Partisi #

Simulasi berikut memodelkan poll loop tiga consumer pada topic enam partisi. Consumer C2 dimatikan pada t=04 untuk pembaruan aplikasi lalu menyala kembali pada t=09 — pola rolling restart yang lazim di Kubernetes. Dua mode menjalani kejadian yang sama persis; satu tick simulasi berarti 100 ms.

"""Simulasi consumer group: rebalance dinamis vs static membership.

Topic 6 partisi, 3 consumer. C2 dimatikan pada t=04 untuk pembaruan aplikasi
lalu menyala kembali pada t=09 (pola rolling restart). 1 tick = 100 ms.
"""

import random

PARTISI = ["P0", "P1", "P2", "P3", "P4", "P5"]
REBALANCE_TICK = 2  # lama stop-the-world setiap rebalance


def tugaskan(aktif):
    """Partisi berurutan dibagi rata ke anggota aktif (pola range assignor)."""
    anggota = sorted(aktif)
    per = len(PARTISI) // len(anggota)
    return {c: PARTISI[i * per:(i + 1) * per] for i, c in enumerate(anggota)}


def format_tugas(tugas):
    return "  ".join(f"{c}={p}" for c, p in tugas.items())


def jalankan(mode):
    rng = random.Random(11)
    tugas = tugaskan(["C1", "C2", "C3"])
    total, henti, tertunda = 0, 0, 0

    def satu_tick(hidup):
        """Consumer hidup memproses partisinya; partisi pemilik mati menunggu."""
        nonlocal total, tertunda
        pesan = 0
        for c, ps in tugas.items():
            beban = sum(rng.randint(2, 5) for _ in ps)
            if c in hidup:
                pesan += beban
            else:
                tertunda += beban  # pesan menumpuk di partisi tanpa pemilik
        total += pesan
        return pesan

    def fase(teks, mulai, jumlah, hidup, jeda=False):
        nonlocal henti
        if jeda:
            henti += jumlah
            print(f"[t={mulai:02d}..{mulai + jumlah - 1:02d}] {teks}: "
                  "0 pesan (stop-the-world)")
            return
        print(f"[t={mulai:02d}..{mulai + jumlah - 1:02d}] {teks}: "
              f"{sum(satu_tick(hidup) for _ in range(jumlah))} pesan")

    print(f"--- mode: {mode} ---")
    print(f"tugas awal : {format_tugas(tugas)}")
    fase("berjalan normal", 1, 3, {"C1", "C2", "C3"})

    print("[t=04] C2 dimatikan untuk pembaruan aplikasi")
    if mode == "dinamis":
        print("[t=04] LeaveGroup diterima coordinator -> REBALANCE #1")
        fase("rebalance", 4, REBALANCE_TICK, set(), jeda=True)
        tugas = tugaskan(["C1", "C3"])
        print(f"tugas baru : {format_tugas(tugas)}")
        fase("C1 dan C3 melayani 6 partisi", 6, 3, {"C1", "C3"})
    else:
        fase("C2 diam, C1 dan C3 tetap memproses", 4, 5, {"C1", "C3"})

    print("[t=09] C2 menyala kembali")
    if mode == "dinamis":
        print("[t=09] C2 bergabung sebagai anggota baru -> REBALANCE #2")
        fase("rebalance", 9, REBALANCE_TICK, set(), jeda=True)
        tugas = tugaskan(["C1", "C2", "C3"])
        print(f"tugas baru : {format_tugas(tugas)}")
        fase("berjalan normal", 11, 3, {"C1", "C2", "C3"})
    else:
        print("coordinator mengenali group.instance.id C2: tugas dipertahankan")
        fase("berjalan normal", 9, 5, {"C1", "C2", "C3"})

    print(f"ringkasan  : {total} pesan, group terhenti {henti} tick "
          f"({henti * 100} ms), pesan tertunda di partisi C2: {tertunda}")
    print()


for mode in ["dinamis", "statis"]:
    jalankan(mode)

Output yang Dihasilkan:

--- mode: dinamis ---
tugas awal : C1=['P0', 'P1']  C2=['P2', 'P3']  C3=['P4', 'P5']
[t=01..03] berjalan normal: 64 pesan
[t=04] C2 dimatikan untuk pembaruan aplikasi
[t=04] LeaveGroup diterima coordinator -> REBALANCE #1
[t=04..05] rebalance: 0 pesan (stop-the-world)
tugas baru : C1=['P0', 'P1', 'P2']  C3=['P3', 'P4', 'P5']
[t=06..08] C1 dan C3 melayani 6 partisi: 61 pesan
[t=09] C2 menyala kembali
[t=09] C2 bergabung sebagai anggota baru -> REBALANCE #2
[t=09..10] rebalance: 0 pesan (stop-the-world)
tugas baru : C1=['P0', 'P1']  C2=['P2', 'P3']  C3=['P4', 'P5']
[t=11..13] berjalan normal: 56 pesan
ringkasan  : 181 pesan, group terhenti 4 tick (400 ms), pesan tertunda di partisi C2: 0

--- mode: statis ---
tugas awal : C1=['P0', 'P1']  C2=['P2', 'P3']  C3=['P4', 'P5']
[t=01..03] berjalan normal: 64 pesan
[t=04] C2 dimatikan untuk pembaruan aplikasi
[t=04..08] C2 diam, C1 dan C3 tetap memproses: 66 pesan
[t=09] C2 menyala kembali
coordinator mengenali group.instance.id C2: tugas dipertahankan
[t=09..13] berjalan normal: 100 pesan
ringkasan  : 230 pesan, group terhenti 0 tick (0 ms), pesan tertunda di partisi C2: 34

Mari kita telaah baris-baris kode di atas:

1. Baris 13-17 (tugaskan): partisi berurutan dibagi rata ke anggota terurut — pola range assignor — sehingga hasil penugasan mudah dibandingkan antar fase.
2. Baris 29-40 (satu_tick): hanya consumer yang hidup memproses partisinya; pesan yang datang di partisi pemilik mati tercatat sebagai tertunda, bukan hilang.
3. Baris 42-50 (fase): tiap fase mencetak rentang tick dan hasilnya; fase dengan jeda=True mencatat lamanya group berhenti total.
4. Baris 56-64 (C2 dimatikan): mode dinamis menerima LeaveGroup lalu rebalance; mode statis tidak menerima pesan keluar apa pun sehingga C1 dan C3 terus bekerja dengan tugas masing-masing.
5. Baris 66-75 (C2 kembali): mode dinamis menjalankan rebalance kedua karena C2 masuk sebagai anggota baru; mode statis cukup dikenali dari group.instance.id dan menerima tugas lamanya kembali.
6. Baris 77-79 (ringkasan): tiga angka penentu keputusan — total pesan, lama group berhenti, dan pesan yang tertunda.

Konsekuensi untuk Pipeline #

Dua ringkasan di output memberi angka yang siap dipakai. Keanggotaan dinamis memproses 181 pesan dengan group terhenti total 400 ms dan tidak ada pesan tertunda; keanggotaan statis memproses 230 pesan tanpa henti sama sekali, dengan 34 pesan menunggu di partisi C2 selama kira-kira lima tick. Pesan tertunda tersebut tidak hilang — pesan-pesan menunggu di partisi hingga pemiliknya kembali — tetapi data pelanggan pada partisi itu menjadi basi selama jendela pembaruan.

Dengan penugasan partisi selesai, pertanyaan berikutnya menyangkut janji pengiriman: apa yang terjadi pada pesan yang sedang diproses ketika consumer berhenti tepat di tengah batch — hilang, atau justru terkirim dua kali? Jawaban sistematisnya menyusul pada Chapter Delivery Semantics.