Key dan partisi

Hashing key dengan murmur2, jaminan pengurutan per partisi, dan strategi memilih key yang sehat.

Daftar isi

Topik ini membahas cara producer menentukan partisi tujuan sebuah pesan, yaitu hashing key dengan algoritma murmur2, jaminan pengurutan yang menyertainya, dan strategi memilih key yang sehat. Hitungan hash diperagakan dalam Python murni tanpa broker, kemudian dibuktikan pada broker nyata dengan producer berkunci. Setelah membacanya, key yang sama dapat dilacak sampai ke partisi yang sama dan urutan pesan per entitas dapat dijaga dengan sengaja.

Pemilihan partisi oleh producer#

Producer menentukan partisi tujuan setiap pesan sebelum pengiriman. Apabila pemanggilan send() menyertakan argumen partisi secara eksplisit, nilai itu yang digunakan. Tanpa argumen partisi, producer memeriksa key pesan, yaitu label opsional yang disertakan bersama nilai pesan. Pesan berkunci diarahkan dengan partitioner, yaitu komponen yang menghitung partisi dari key, sedangkan pesan tanpa key disebar antar partisi oleh sticky partitioner yang menetap di satu partisi selama satu batch lalu pindah ke partisi berikutnya.

Rumus partitioner default untuk pesan berkunci sederhana. Key berupa bytes dihitung dengan hash murmur2, lalu hasilnya disamarkan menjadi bilangan non-negatif dan diambil sisanya setelah dibagi dengan jumlah partisi, dalam bentuk partisi = (murmur2(key) & 0x7fffffff) % jumlah_partisi. Murmur2 adalah fungsi hash non-kriptografis yang cepat dan deterministik, sehingga key yang sama selalu menghasilkan partisi yang sama pada jumlah partisi yang tetap. Nilai hashnya identik di seluruh klien, sehingga pesan dengan key yang sama berakhir di partisi yang sama baik dikirim oleh klien Java maupun Python. Sifat inilah yang menjadi dasar jaminan pengurutan per entitas pada bagian berikutnya.

Menghitung hash murmur2 dalam Python#

Skrip berikut mengimplementasikan murmur2 tanpa pustaka apa pun, persis sebagaimana digunakan partitioner default Kafka. Fungsi tersebut memproses masukan empat byte sekaligus dengan konstanta peracikan yang tetap, menangani sisa byte di akhir satu per satu, lalu mengacak hasil dengan dua langkah penutupan. Operasi masking dengan 0xFFFFFFFF menjaga hasil tetap dalam 32 bit, sedangkan mask 0x7FFFFFFF pada fungsi pemetaan menjadikan nilai selalu non-negatif sebelum modulo.

python murmur2-demo.py

def murmur2(data):
    """Hash murmur2 sebagaimana digunakan default partitioner Kafka."""
    length = len(data)
    seed = 0x9747b28c
    m = 0x5bd1e995
    r = 24
    h = (seed ^ length) & 0xFFFFFFFF
    for i in range(length // 4):
        i4 = i * 4
        k = (data[i4] & 0xFF) + ((data[i4 + 1] & 0xFF) << 8) \
            + ((data[i4 + 2] & 0xFF) << 16) + ((data[i4 + 3] & 0xFF) << 24)
        k = (k * m) & 0xFFFFFFFF
        k ^= k >> r
        k = (k * m) & 0xFFFFFFFF
        h = (h * m) & 0xFFFFFFFF
        h ^= k
    sisa = length % 4
    awal = length & ~3
    if sisa == 3:
        h ^= (data[awal + 2] & 0xFF) << 16
        h ^= (data[awal + 1] & 0xFF) << 8
        h ^= data[awal] & 0xFF
        h = (h * m) & 0xFFFFFFFF
    elif sisa == 2:
        h ^= (data[awal + 1] & 0xFF) << 8
        h ^= data[awal] & 0xFF
        h = (h * m) & 0xFFFFFFFF
    elif sisa == 1:
        h ^= data[awal] & 0xFF
        h = (h * m) & 0xFFFFFFFF
    h ^= h >> 13
    h = (h * m) & 0xFFFFFFFF
    h ^= h >> 15
    return h


def partisi_untuk(key, jumlah_partisi):
    """Peta key ke nomor partisi dengan rumus default partitioner Kafka."""
    return (murmur2(key) & 0x7FFFFFFF) % jumlah_partisi


kunci = ["kunci-1", "kunci-2", "kunci-3"]
for nama in kunci:
    data = nama.encode()
    print(f"{nama}: hash={murmur2(data)} partisi(3)={partisi_untuk(data, 3)}")

# key yang sama dihitung ulang menghasilkan hash dan partisi yang sama
data = "kunci-1".encode()
print(f"uji ulang kunci-1: hash={murmur2(data)} partisi(3)={partisi_untuk(data, 3)}")

text Keluaran

kunci-1: hash=3443998164 partisi(3)=1
kunci-2: hash=2468232559 partisi(3)=2
kunci-3: hash=4217351285 partisi(3)=0
uji ulang kunci-1: hash=3443998164 partisi(3)=1

Baris uji ulang menunjukkan hash dan partisi yang persis sama untuk key yang sama, karena murmur2 tidak memiliki komponen acak sama sekali. Implementasi ini telah dibandingkan dengan fungsi murmur2 pada pustaka kafka-python untuk sejumlah key dan hasilnya identik. Nilai partisi pada keluaran ini digunakan kembali oleh demo broker pada bagian berikut, sehingga perhitungan murni dan keputusan broker dapat dibandingkan secara langsung.

Jaminan pengurutan per partisi#

Pengurutan hanya dijamin di dalam satu partisi karena setiap partisi adalah log kejadian tersendiri dengan urutan offsetnya sendiri. Demo berikut mengirim enam pesan berkunci ke topic tiga partisi, dengan tiga key yang masing-masing muncul dua kali. Metadata setiap pengiriman dicetak, lalu seluruh pesan dibaca kembali dan diurutkan menurut partisi dan offset agar urutan di dalam setiap partisi terlihat jelas.

python key-demo.py

from kafka import KafkaProducer, KafkaConsumer, KafkaAdminClient
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError
import time

topic = f"m1-key-{int(time.time())}"  # nama unik per eksekusi

admin = KafkaAdminClient(bootstrap_servers="localhost:9092")
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=3, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()

producer = KafkaProducer(bootstrap_servers="localhost:9092")
urutan = ["kunci-1", "kunci-2", "kunci-3", "kunci-1", "kunci-2", "kunci-3"]
for k in urutan:
    info = producer.send(topic, key=k.encode(), value=f"pesan-{k}".encode()).get(timeout=10)
    print(f"kirim key={k} -> partisi={info.partition} offset={info.offset}")
producer.flush()
producer.close()

consumer = KafkaConsumer(
    topic,
    bootstrap_servers="localhost:9092",
    auto_offset_reset="earliest",
)
pesan = []
while len(pesan) < 6:
    batch = consumer.poll(timeout_ms=5000, max_records=6 - len(pesan))
    if not batch:
        break
    for daftar in batch.values():
        pesan.extend(daftar)
# pesan dari tiga partisi tiba berlapis, sehingga diurutkan menurut
# partisi dan offset agar urutan di dalam setiap partisi terlihat jelas
pesan.sort(key=lambda m: (m.partition, m.offset))
for msg in pesan:
    print(f"terima partisi={msg.partition} offset={msg.offset} key={msg.key.decode()}")
consumer.close()

text Keluaran

kirim key=kunci-1 -> partisi=1 offset=0
kirim key=kunci-2 -> partisi=2 offset=0
kirim key=kunci-3 -> partisi=0 offset=0
kirim key=kunci-1 -> partisi=1 offset=1
kirim key=kunci-2 -> partisi=2 offset=1
kirim key=kunci-3 -> partisi=0 offset=1
terima partisi=0 offset=0 key=kunci-3
terima partisi=0 offset=1 key=kunci-3
terima partisi=1 offset=0 key=kunci-1
terima partisi=1 offset=1 key=kunci-1
terima partisi=2 offset=0 key=kunci-2
terima partisi=2 offset=1 key=kunci-2

Broker menempatkan kunci-1 di partisi 1, kunci-2 di partisi 2, dan kunci-3 di partisi 0, persis seperti hitungan partisi(3) pada demo murmur2. Pemetaan ini berlaku tanpa pengecualian, sehingga pesan kedua dengan key yang sama mendapat offset berikutnya di partisi yang sama. Kelompok terima memperlihatkan offset yang menaik per partisi, yaitu jaminan urutan, sedangkan antar partisi tidak ada urutan yang dapat dibandingkan.

Memilih key yang sehat#

Key menempel pada pesan selama pesan itu tersimpan di log, sehingga pemilihan key menyerupai keputusan desain dan bukan detail pemanggilan. Key yang sehat memenuhi tiga kriteria. Pertama, kardinalitasnya cukup tinggi, misalnya id sensor atau id perangkat, agar partisi dapat terisi merata. Kedua, sebarannya merata tanpa key yang mendominasi, karena satu key yang sangat sibuk membuat satu partisi menanggung hampir seluruh beban, keadaan yang disebut hot key. Ketiga, nilainya stabil sepanjang waktu, karena partisi tujuan dihitung dari key saat pesan dikirim.

Key yang dihasilkan dari waktu, seperti UUID versi waktu, sebaiknya dihindari apabila urutan per entitas dibutuhkan, karena nilai yang selalu berbeda membuat pesan-pesan satu entitas tersebar tanpa jaminan urutan. Sebaliknya, pesan tanpa key tetap sah untuk aliran yang tidak menuntut urutan antar pesan, dan sticky partitioner akan menyebarkannya antar partisi. Keputusannya bergantung pada kebutuhan, yaitu key digunakan ketika kejadian satu entitas harus dibaca berurutan.

Ringkasan#

Nomor partisi pesan berkunci dihitung dengan hash murmur2 lalu modulo jumlah partisi, sehingga key yang sama selalu berakhir di partisi yang sama.
Implementasi murmur2 pada topik ini menghasilkan nilai yang identik dengan pustaka klien, dan demo broker mengonfirmasi hasil perhitungannya.
Pengurutan dijamin hanya di dalam satu partisi, dan key yang konsisten menjaga urutan per entitas.
Key sehat memiliki kardinalitas tinggi, sebaran yang merata, dan nilai yang stabil sepanjang waktu.

Referensi#

Producer configs — Apache Software Foundation — deskripsi partitioner.class yang menjelaskan pemilihan partisi dari hash key dan sticky partitioner untuk pesan tanpa key.
Apache Kafka design — Apache Software Foundation — catatan desain resmi tentang penugasan partisi dan jaminan pengurutan pada consumer group.
kafka/partitioner/default.py — dpkp/kafka-python — implementasi murmur2 pada pustaka kafka-python yang digunakan untuk membandingkan hasil hash pada topik ini.