Bagian 5: Mengatur Topic dan Partition

Jumlah partition menentukan batas paralelisme, tetapi yang menentukan beban tiap partition adalah kunci pesan. Producer yang mengirim semua pesanan berkunci wilayah akan menumpuk 70% traffic TokoKita pada satu partition — persis seperti hot key pada chapter Partitioning dan Sharding — dan consumer pemilik partition itu bekerja jauh lebih berat daripada rekan satu group-nya. Bagian ini membedah partitioner murmur2 yang dipakai klien Kafka resmi, memverifikasi implementasinya terhadap vektor uji resmi, lalu mengukur dua strategi kunci pada topic tokokita.pesanan dengan 120 event sungguhan dari simulator.

Partitioner murmur2 #

Klien Kafka menghitung partition pesan berkunci dengan hash murmur2: hasil hash dijadikan positif lalu diambil modulo jumlah partition. Fungsi berikut adalah port dari org.apache.kafka.common.utils.Utils.murmur2 pada src/partitioner.py:

def murmur2(data: bytes) -> int:
    """Hash murmur2 32-bit atas ``data``; hasil bertanda seperti ``int`` Java.

    Nilai kembalian bisa negatif karena Java memakai bilangan bertanda
    32-bit; pemanggil harus menormalkan lewat :func:`to_positive` sebelum
    dipakai sebagai indeks partition.
    """
    panjang = len(data)
    seed = 0x9747B28C
    m = 0x5BD1E995
    r = 24
    h = (seed ^ panjang) & 0xFFFFFFFF
    panjang4 = panjang // 4

    for i in range(panjang4):
        i4 = i * 4
        k = data[i4] | (data[i4 + 1] << 8) | (data[i4 + 2] << 16) | (data[i4 + 3] << 24)
        k = (k * m) & 0xFFFFFFFF
        k ^= k >> r  # geser tak bertanda di Java; k di sini sudah non-negatif
        k = (k * m) & 0xFFFFFFFF
        h = (h * m) & 0xFFFFFFFF

Pemetaan akhir ke partition berada di fungsi kecil berikut:

def partisi_dari_kunci(kunci: str | bytes, jumlah_partisi: int) -> int:
    """Pilih partition untuk ``kunci`` (murmur2 positif modulo jumlah partisi)."""
    if jumlah_partisi <= 0:
        raise ValueError("jumlah_partisi harus lebih besar dari 0")
    data = kunci.encode("utf-8") if isinstance(kunci, str) else kunci
    return to_positive(murmur2(data)) % jumlah_partisi

Mari kita telaah baris-baris kode di atas (nomor mengikuti posisi di src/partitioner.py):

1. Baris 21-23 (seed, m, r): konstanta algoritma murmur2 — tidak boleh diubah, karena nilai hash harus sama dengan klien Java agar kunci yang sama mendarat di partition yang sama lintas bahasa.
2. Baris 24 (h = (seed ^ panjang) & ...): state hash diawali dari panjang data; operasi 32-bit diselubungi & 0xFFFFFFFF karena Python punya bilangan besar tanpa overflow, sedangkan Java int membungkus di 32 bit.
3. Baris 27-33 (loop per blok 4 byte): data diproses empat byte sekaligus; setiap blok diacak dengan perkalian m dan geser r, lalu dicampurkan ke h — inti pengacakan murmur2.
4. Baris 61-66 (partisi_dari_kunci): kunci string dienkode UTF-8 dulu, hash dijadikan positif lewat to_positive (operasi & 0x7fffffff), lalu modulo jumlah partition menentukan tujuan akhir.

Implementasi ini diverifikasi terhadap vektor uji resmi UtilsTest Apache Kafka — misalnya murmur2(b"21") == -973932308 dan murmur2(b"foobar") == -790332482 — dan tes test_murmur2_vektor_resmi_kafka mengunci enam vektor tersebut. Artinya, kunci "jakarta" mendarat di partition yang sama di simulasi kita dan di cluster Kafka yang menjalankan partitioner bawaan.

Kunci Menentukan Beban #

Simulator checkout memilih wilayah dengan bobot timpang — jakarta 70, bandung 10, surabaya 8, medan 7, makassar 5 — meniru toko daring yang memang terpusat di satu kota. Skrip berikut membandingkan dua strategi kunci pada 120 event yang sama:

from src.local_kafka import LocalKafka
from src.partitioner import partisi_dari_kunci
from src.producer import BOBOT_WILAYAH, WILAYAH, SimulatorCheckout, buat_producer

print("pemetaan kunci wilayah -> partition (murmur2, 3 partition):")
for wilayah in WILAYAH:
    print(f"  {wilayah:9s} -> p{partisi_dari_kunci(wilayah, 3)}")

print(f"\nbobot wilayah simulator: {dict(zip(WILAYAH, BOBOT_WILAYAH))}")

broker = LocalKafka()
broker.create_topic("tokokita.pesanan", num_partitions=3)
producer = buat_producer(broker=broker)
SimulatorCheckout(producer, kunci="wilayah", bobot_wilayah=BOBOT_WILAYAH).kirim(120)
print("\ndistribusi pesan per partition (kunci = wilayah, timpang):")
for stat in broker.statistik_topic("tokokita.pesanan"):
    print(f"  partition {stat['partition']}: {stat['jumlah_pesan']} pesan")

broker2 = LocalKafka()
broker2.create_topic("tokokita.pesanan", num_partitions=3)
producer2 = buat_producer(broker=broker2)
SimulatorCheckout(producer2, kunci="pesanan_id", bobot_wilayah=BOBOT_WILAYAH).kirim(120)
print("distribusi pesan per partition (kunci = pesanan_id, beragam):")
for stat in broker2.statistik_topic("tokokita.pesanan"):
    print(f"  partition {stat['partition']}: {stat['jumlah_pesan']} pesan")

Output yang Dihasilkan:

pemetaan kunci wilayah -> partition (murmur2, 3 partition):
  jakarta   -> p1
  bandung   -> p0
  surabaya  -> p0
  medan     -> p2
  makassar  -> p0

bobot wilayah simulator: {'jakarta': 70, 'bandung': 10, 'surabaya': 8, 'medan': 7, 'makassar': 5}

distribusi pesan per partition (kunci = wilayah, timpang):
  partition 0: 31 pesan
  partition 1: 81 pesan
  partition 2: 8 pesan
distribusi pesan per partition (kunci = pesanan_id, beragam):
  partition 0: 39 pesan
  partition 1: 41 pesan
  partition 2: 40 pesan

Hasilnya tegas. Kunci wilayah menumpuk 81 dari 120 pesan (68%) di partition 1 karena jakarta — pemegang bobot 70 — meng-hash ke p1; partition 2 hanya menerima 8 pesan dari medan. Ganti kunci ke pesanan_id yang kardinalitasnya tinggi, murmur2 menyebar 39/41/40 hampir sempurna. Perhatikan juga bahwa hash hanya menentukan tujuan, bukan keadilan: murmur2 tidak tahu bahwa “bandung, surabaya, makassar kebetulan satu partition” — ketidakadilan itu murni karena jumlah kunci (5) terlalu sedikit untuk 3 partition.

Memilih Kunci Topic #

Pilihan kunci adalah tukar-menukar antara jaminan urutan dan keadilan beban:

Untuk tokokita.pesanan, proyek ini memakai pelanggan_id sebagai kunci bawaan: 199 pelanggan simulator tersebar cukup merata (eksekusi nyata 120 event menghasilkan 44/35/41), dan urutan event per pelanggan terjaga — berguna bila hilir ingin tahu urutan status pesanan dari pelanggan yang sama.

Dengan topic yang beban terbagi, kita bisa fokus pada pertanyaan berikutnya: bagaimana consumer group menandai posisi bacanya, dan apa yang terjadi pada pesan ketika consumer berhenti di tengah jalan — urusan offset pada bagian selanjutnya.