Producers dan Consumers

Pada arsitektur streaming, aplikasi yang membuat data dan aplikasi yang membaca data tidak pernah berkomunikasi langsung. Layanan checkout TokoKita cukup menulis setiap pesanan sebagai event ke broker, lalu layanan notifikasi, deteksi penipuan, dan pipeline analitik membacanya masing-masing. Komponen penulis disebut producer, komponen pembaca disebut consumer, dan keduanya hanya dihubungkan oleh topic yang sama. Chapter ini membedah mekanika producer sampai konfigurasi acks dan idempotent producer, menutup dengan poll loop consumer, lalu dipertegas simulasi Python murni tanpa broker.

Producer: Dari produce() ke Record Batch #

Pemanggilan produce() tidak langsung mengirim apa pun ke jaringan. Pustaka klien terlebih dahulu menjalankan serializer atas value, lalu partitioner memilih partisi tujuan — pesanan dengan key user-1 selalu jatuh ke partisi yang sama, menjaga urutan per pelanggan (anatomi topic dan partisi dibedah pada Chapter Topics, Partitions, dan Offsets). Pesan lalu menumpuk di record batch per partisi; pengiriman terjadi saat batch penuh atau batas linger.ms tercapai — cara Kafka menukar sedikit latensi dengan throughput besar.

Karena pengiriman bersifat asinkron, produce() kembali segera; nasib pesan sesungguhnya dilaporkan belakangan lewat delivery report callback. Callback inilah satu-satunya tempat kode kita mengetahui partisi, offset, atau galatnya.

Konfigurasi acks dan Idempotent Producer #

Janji penyimpanan ditentukan oleh satu konfigurasi: berapa banyak pihak yang harus menulis sebelum broker menjawab “tersimpan”.

Producer modern Kafka memakai acks=all sebagai nilai bawaan (dokumentasi Apache Kafka ); nilai lain adalah pilihan sadar untuk menukar daya tahan dengan latensi. Satu masalah tetap tinggal: ack yang hilang membuat producer tidak tahu apakah pesannya sampai, sehingga producer mengirim ulang dan broker mendapat dua salinan. Idempotent producer memutus lingkaran ini: setiap pesan membawa nomor urut per partisi, dan broker membuang tulisan ulang dengan nomor urut yang sudah pernah diterimanya.

Producer Python dengan confluent-kafka #

Kode berikut mengikuti API pustaka confluent-kafka; sintaksnya diverifikasi dengan py_compile dan tidak dijalankan di lingkungan ini karena tidak ada broker — perilakunya dijelaskan dalam prosa.

"""Contoh producer Kafka (API confluent-kafka): tidak dijalankan di sini."""

from confluent_kafka import Producer


def laporan_pengiriman(err, msg):
    """Callback yang dipanggil saat broker menjawab (atau menolak)."""
    if err is not None:
        print(f"Gagal: {err}")
    else:
        print(f"Terkirim ke {msg.topic()}[{msg.partition()}] offset {msg.offset()}")


producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "acks": "all",                 # 0, 1, atau all
    "enable.idempotence": "true",  # broker membuang duplikat retry
    "linger.ms": "5",              # tunggu 5 ms agar batch terisi penuh
})

for nomor in range(10):
    producer.produce(
        "pesanan",
        key=f"user-{nomor % 3}",
        value=f'{{"id": {nomor}, "total": 250000}}',
        on_delivery=laporan_pengiriman,
    )

producer.flush(10)

Konfigurasi acks dan enable.idempotence memilih janji pengiriman, linger.ms mengatur lamanya batch dikumpulkan, dan key user-{nomor % 3} membuat pesanan tiga pelanggan tetap terurut per partisi. flush(10) memaksa producer mengirim sisa batch dan menunggu callback paling lama 10 detik.

Simulasi: Ack Hilang, Retry, dan Duplikat #

Simulasi berikut memodelkan inti masalah di atas dengan Python murni: broker selalu berhasil menulis, tetapi 15% jawabannya hilang di jaringan sehingga producer timeout dan mengirim ulang.

"""Simulasi mini producer: pengaruh acks dan idempotence."""

import random

JUMLAH_PESAN = 200
PELUANG_ACK_HILANG = 0.15   # 15% jawaban broker tidak sampai ke producer
PELUANG_GAGAL_TULIS = 0.10  # 10% tulisan broker gagal (hanya berlaku acks=0)
MAKS_RETRY = 5


def kirim(acks, idempotent, seed=42):
    """Kirim JUMLAH_PESAN pesan; kembalikan (unik, duplikat, hilang, rata-tunggu)."""
    rng = random.Random(seed)
    unik, duplikat, hilang = 0, 0, 0
    total_tunggu = 0.0

    for _ in range(JUMLAH_PESAN):
        if acks == "0":
            # fire-and-forget: tidak menunggu jawaban apa pun
            if rng.random() < PELUANG_GAGAL_TULIS:
                hilang += 1          # producer tidak pernah tahu
            else:
                unik += 1
            continue

        for _upaya in range(1, MAKS_RETRY + 1):
            # broker menulis lalu menjawab; all menunggu replika -> lebih lama
            total_tunggu += rng.uniform(2, 4) if acks == "1" else rng.uniform(8, 14)
            if rng.random() >= PELUANG_ACK_HILANG:
                unik += 1            # ack diterima: pesan dianggap sampai
                break
            # ack timeout -> kirim ulang
            if not idempotent:
                duplikat += 1        # upaya sebelumnya ternyata sudah tertulis

    return unik, duplikat, hilang, total_tunggu / JUMLAH_PESAN


print(f"Pesan dikirim : {JUMLAH_PESAN}")
print(f"Jaringan      : {PELUANG_ACK_HILANG:.0%} ack hilang, "
      f"{PELUANG_GAGAL_TULIS:.0%} tulisan broker gagal\n")
print(f"{'Mode':<28}{'Unik':>6}{'Duplikat':>10}{'Hilang':>8}{'Tunggu/pesan':>14}")

for label, acks, idem in [
    ("acks=0", "0", False),
    ("acks=1 tanpa idempotence", "1", False),
    ("acks=1 idempotent", "1", True),
    ("acks=all idempotent", "all", True),
]:
    unik, dup, hil, rata = kirim(acks, idem)
    print(f"{label:<28}{unik:>6}{dup:>10}{hil:>8}{rata:>11.1f} ms")

Output yang Dihasilkan:

Pesan dikirim : 200
Jaringan      : 15% ack hilang, 10% tulisan broker gagal

Mode                          Unik  Duplikat  Hilang  Tunggu/pesan
acks=0                         176         0      24        0.0 ms
acks=1 tanpa idempotence       200        29       0        3.5 ms
acks=1 idempotent              200         0       0        3.5 ms
acks=all idempotent            200         0       0       12.7 ms

Mari kita telaah baris-baris kode di atas:

1. Baris 5-8 (konstanta model): jaringan yang sama dihadapi semua mode — 15% ack hilang, 10% tulisan broker gagal — sehingga perbandingan antar baris output adil.
2. Baris 18-24 (cabang acks=0): producer tidak menunggu jawaban; tulisan yang gagal berakhir sebagai hilang tanpa pernah diketahui, dan retry tidak pernah terjadi.
3. Baris 28 (waktu tunggu): mode all menunggu replika sehingga waktunya 8-14 ms, sementara acks=1 cukup 2-4 ms.
4. Baris 29-31 (ack diterima): pesan dihitung unik dan producer berhenti mencoba.
5. Baris 33-34 (retry tanpa idempotence): upaya pertama ternyata sudah tertulis di broker, jadi setiap retry menambah satu duplikat.
6. Baris 44-48 (empat mode): dua baris acks=1 menghadapi urutan kejadian yang sama persis karena seed-nya sama — satu-satunya perbedaan hasil mereka adalah idempotence.

Perhatikan dua baris acks=1: dengan urutan kejadian jaringan yang sama persis, 29 duplikat berubah menjadi nol hanya karena broker mengenali nomor urut. Harga acks=all terlihat jelas: 12,7 ms berbanding 3,5 ms per pesan — sekitar tiga setengah kali lipat, tarif yang layak untuk data pesanan.

Consumer: Poll Loop dan Offset Commit #

Sisi pembaca bekerja dengan pola tarik, bukan dorong. Consumer berlangganan topic, lalu menjalankan poll loop: poll() mengembalikan sekelompok pesan dari partisi miliknya, kode kita memprosesnya, dan posisi baca (offset) maju. Yang membuat pola ini penting bagi data engineer:

1. Consumer sendiri yang menentukan laju baca — broker tidak mendorong apa pun, sehingga consumer yang lambat tidak menyeret broker.
2. Posisi baca di-commit ke internal topic __consumer_offsets, sehingga consumer boleh berhenti dan menyala ulang tanpa kehilangan tempat.
3. Commit berkala terjadi otomatis; kontrol penuh diperoleh dengan mematikan auto-commit dan meng-commit offset sendiri setelah hasil pemrosesan aman.
4. Jarak antara offset terkomit dan offset terbaca (consumer lag) adalah metrik kesehatan pipeline yang paling sering dipantau.

Detail konsekuensi commit — kapan pesan hilang, kapan pesan dobel — adalah materi Chapter Delivery Semantics. Sebelum ke sana, satu pertanyaan harus dijawab dulu: satu consumer saja tidak cukup mengejar jutaan pesanan, dan cara consumer bekerja sama dalam satu group dibedah pada Chapter Consumer Groups.