KAL
Menu

Consumer group

Koordinasi consumer group: distribusi partisi ke anggota, pembagian beban, dan identitas group.

Daftar isi

Topik ini membahas cara sekumpulan consumer bekerja sebagai satu group, mulai dari identitas group yang menyatukan anggota, distribusi partisi kepada anggota, sampai pembacaan tabel deskripsi group pada CLI. Penugasan partisi diperagakan lewat simulasi assignor dalam Python murni sehingga hasilnya dapat dihitung ulang, sedangkan koordinasi dan posisi baca diuji pada broker nyata menggunakan consumer console. Setelah membacanya, pembagian beban antar anggota group dapat dijelaskan dari mekanisme penugasannya, bukan sekadar dari hasil pengamatan.

Identitas group dan koordinasi#

Consumer group adalah sekumpulan consumer yang berbagi satu identitas, yaitu group id yang diatur lewat konfigurasi group.id pada setiap anggotanya. Anggota dapat berupa beberapa proses pada satu mesin maupun instance aplikasi yang tersebar di beberapa mesin. Bagi broker, seluruh consumer dengan group.id yang sama adalah satu kesatuan yang membagi pekerjaan konsumsi topic yang sama.

Pengatur keanggotaan dipegang oleh group coordinator, yaitu broker yang ditugaskan mengawasi satu group. Setiap anggota mendaftar lewat permintaan join group, lalu coordinator memilih salah satu anggota sebagai pemimpin penugasan (leader). Anggota pemimpin menyusun pembagian partisi menurut assignor yang dikonfigurasikan, kemudian hasilnya disebar kembali kepada seluruh anggota lewat balasan sync group. Ketika satu anggota keluar atau anggota baru mendaftar, keanggotaan berubah dan penugasan dihitung ulang, proses yang disebut rebalancing.

Posisi baca dicatat per pasangan group dan partisi, sehingga dua group yang membaca topic yang sama berjalan saling tidak memengaruhi. Masing-masing group menerima seluruh pesan sesuai posisi bacanya sendiri tanpa menggeser offset group lain. Sifat ini membuat group cocok untuk dua keperluan yang berbeda, yaitu membagi beban di dalam satu aplikasi dan menjalankan beberapa aplikasi yang membutuhkan aliran data yang sama.

Distribusi partisi ke anggota#

Satuan pembagian pekerjaan di dalam satu group adalah partisi, bukan pesan. Satu partisi dari satu topic hanya boleh dibaca satu anggota pada satu waktu, sedangkan satu anggota boleh memegang beberapa partisi sekaligus. Jaminan satu pemilik per partisi itulah yang membuat posisi baca per partisi dapat dikelola tanpa koordinasi tambahan di sisi anggota.

Komponen yang menyusun pembagian disebut assignor dan dipilih lewat konfigurasi partition.assignment.strategy pada klien. Assignor range mengurutkan partisi lalu membagikannya berurutan kepada anggota yang juga terurut, sehingga kelebihan partisi dipikul oleh anggota pertama apabila jumlah partisi tidak habis dibagi. Assignor roundrobin membagikannya bergiliran satu per satu, sehingga kelebihan partisi tersebar lebih merata. Keduanya mudah dihitung ulang dengan tangan, sedangkan klien modern menyediakan pilihan tambahan yang mempertahankan penugasan lama selama keanggotaan tidak berubah.

Simulasi berikut mengimplementasikan kedua assignor tersebut dalam Python murni tanpa broker, sehingga hasilnya dapat dibandingkan langsung dari kodenya.

python simulasi-assignor.py 
def assign_range(partisi, anggota):
    """Assignor range: partisi satu topic dibagi berurutan ke anggota terurut."""
    penugasan = {a: [] for a in anggota}
    jumlah = len(partisi) // len(anggota)
    sisa = len(partisi) % len(anggota)
    awal = 0
    for i, a in enumerate(anggota):
        panjang = jumlah + (1 if i < sisa else 0)
        penugasan[a] = partisi[awal:awal + panjang]
        awal += panjang
    return penugasan


def assign_roundrobin(partisi, anggota):
    """Assignor roundrobin: partisi dibagikan bergiliran ke seluruh anggota."""
    penugasan = {a: [] for a in anggota}
    for i, p in enumerate(partisi):
        penugasan[anggota[i % len(anggota)]].append(p)
    return penugasan


partisi = [0, 1, 2, 3, 4]

print("range, dua anggota:")
for a, p in assign_range(partisi, ["consumer-a", "consumer-b"]).items():
    print(f"  {a}: {p}")

print("range, tiga anggota:")
for a, p in assign_range(partisi, ["consumer-a", "consumer-b", "consumer-c"]).items():
    print(f"  {a}: {p}")

print("roundrobin, tiga anggota:")
for a, p in assign_roundrobin(partisi, ["consumer-a", "consumer-b", "consumer-c"]).items():
    print(f"  {a}: {p}")
text Keluaran 
range, dua anggota:
  consumer-a: [0, 1, 2]
  consumer-b: [3, 4]
range, tiga anggota:
  consumer-a: [0, 1]
  consumer-b: [2, 3]
  consumer-c: [4]
roundrobin, tiga anggota:
  consumer-a: [0, 3]
  consumer-b: [1, 4]
  consumer-c: [2]

Keluaran memperlihatkan perbedaan kedua assignor pada topic lima partisi. Dengan range dan dua anggota, consumer-a menerima tiga partisi pertama sedangkan consumer-b menerima dua sisanya, karena sisa pembagian diberikan kepada anggota yang berada lebih dulu dalam urutan. Dengan tiga anggota, kelebihan satu partisi jatuh kepada consumer-a. Roundrobin justru menaruh partisi 1 dan 4 pada consumer-b, sehingga tiga anggota memegang dua, dua, dan satu partisi dengan sebaran yang berselang-seling. Diagram berikut merangkum penugasan range untuk kasus dua anggota.

Topic, 5 partisipartisi 0partisi 1partisi 2partisi 3partisi 4Consumer groupconsumer-aconsumer-b
Assignor range membagi lima partisi ke dua anggota group

Konsumsi bersama pada CLI#

Consumer console dapat dijalankan sebagai anggota group dengan opsi --group, sehingga koordinasi group dapat diuji tanpa menulis program. Demo ini menggunakan topic tetap bernama m1-grup-demo-t yang selalu dihapus dan dibuat ulang pada setiap eksekusi, sehingga jumlah partisi dan isi lognya selalu sama. Blok berikut menyiapkan topic tersebut lalu mengisinya dengan tiga pesan.

bash 
(
  cd /home/kal/my-project/kafka-runtime/kafka
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --if-exists --topic m1-grup-demo-t 2>/dev/null
  sleep 2
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic m1-grup-demo-t --partitions 1
  printf "satu\ndua\ntiga\n" | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic m1-grup-demo-t 2>/dev/null
)
text Keluaran 
Created topic m1-grup-demo-t.

Blok berikut menjalankan consumer console sebagai anggota group m1-grup-demo. Opsi --from-beginning mengarahkan anggota baru yang belum memiliki posisi tersimpan untuk membaca dari awal log, sedangkan --max-messages 3 membuat anggota berhenti sendiri setelah kuota tiga pesan terpenuhi. Consumer console mencetak setiap pesan yang berhasil dibacanya, kemudian perintah kafka-consumer-groups --describe menampilkan tabel posisi baca group tersebut.

bash 
(
  cd /home/kal/my-project/kafka-runtime/kafka
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic m1-grup-demo-t --group m1-grup-demo --from-beginning --max-messages 3 2>/dev/null
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group m1-grup-demo 2>/dev/null
)
text Keluaran 
satu
dua
tiga

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
m1-grup-demo    m1-grup-demo-t  0          3               3               0               -               -               -

Tiga baris pertama pada keluaran adalah pesan yang dibaca consumer, sedangkan tabel di bawahnya memperlihatkan posisi baca group dengan satu baris per partisi. Kolom CURRENT-OFFSET adalah posisi baca tercatat, LOG-END-OFFSET adalah ujung log partisi, dan LAG adalah selisih keduanya, yaitu jumlah pesan yang belum dibaca. Nilai LAG 0 menunjukkan seluruh pesan sudah terbaca, sedangkan nilai - pada CONSUMER-ID, HOST, dan CLIENT-ID berarti satu-satunya anggota sudah keluar setelah kuota pesannya terpenuhi. Group yang seluruh anggotanya keluar tetap tercatat di broker beserta posisi bacanya, sehingga groupnya tetap dapat dideskripsikan. Pada group yang anggotanya masih berjalan, ketiga kolom tersebut terisi nama anggota beserta host-nya.

Skalabilitas dan batas group#

Kapasitas konsumsi satu group dibatasi oleh jumlah partisi topic yang dibacanya. Menambah anggota membagi partisi lebih tipis, sehingga beban per anggota turun sampai setiap anggota memegang satu partisi. Anggota berikutnya tidak memperoleh partisi apa pun dan hanya menambah beban koordinasi, sehingga disebut anggota idle. Anggota idle tetap tercatat sebagai anggota group, tetapi tidak memperoleh pembagian pekerjaan apa pun sampai jumlah partisi bertambah atau salah satu anggota lain keluar.

Batas atas paralelisme tersebut membuat jumlah partisi menjadi keputusan yang berdampak pada group sejak awal. Topic lima partisi menampung paling banyak lima anggota yang aktif membaca, dan jumlah anggota yang melebihi itu tidak menambah kapasitas konsumsi apa pun. Demo CLI di topik ini menggunakan topic satu partisi, sehingga group m1-grup-demo menampung satu anggota aktif saja, dan anggota kedua yang bergabung tidak memperoleh partisi apa pun. Sebaliknya, group dengan satu anggota mengolah seluruh partisi sendirian, sehingga kegagalan satu proses menghentikan konsumsi sampai anggota lain bergabung dan penugasan dihitung ulang. Karena pekerjaan dibagi per partisi, throughput group terakumulasi dari throughput per partisi setiap anggotanya.

Kesalahan umum#

  1. Dua aplikasi berbeda menggunakan group.id yang sama sehingga pesan terbagi di antara keduanya tanpa disadari. Masing-masing aplikasi hanya menerima sebagian pesan, dan tabel deskripsi group menampilkan anggota dari dua aplikasi sekaligus. Perbaikannya dengan memberi satu group.id khusus untuk setiap aplikasi, sedangkan aplikasi yang sama-sama membutuhkan seluruh pesan menggunakan group terpisah.
  2. Jumlah anggota yang dijalankan melebihi jumlah partisi sehingga sebagian anggota tidak memperoleh partisi. Consumer tersebut tampak hidup tetapi tidak pernah membaca pesan apa pun. Sejajarkan jumlah anggota dengan jumlah partisi, atau tambah partisi topic apabila paralelisme lebih tinggi memang dibutuhkan.
  3. Consumer CLI milik group yang sudah memiliki posisi tersimpan dijalankan dengan opsi --from-beginning, tetapi keluarannya tetap kosong. Opsi --from-beginning hanya digunakan apabila group belum memiliki posisi tersimpan untuk partisi terkait, sedangkan posisi yang sudah ada selalu dilanjutkan. Untuk membaca ulang dari awal, atur ulang posisi group dengan opsi --reset-offsets pada kafka-consumer-groups.sh.

Ringkasan#

  • Consumer group menyatukan anggota lewat group.id, dan posisi baca dicatat per pasangan group dan partisi.
  • Satu partisi hanya dibaca satu anggota di dalam satu group, sehingga pembagian beban berarti pembagian partisi.
  • Assignor range membagi partisi berurutan dan roundrobin membagikannya bergiliran, dengan hasil yang dapat dihitung ulang.
  • Tabel --describe menampilkan posisi baca dan lag per partisi, sedangkan anggota yang sudah keluar ditandai -.

Referensi#