Delivery Semantics

Antara “pesan dibaca dari topic” dan “hasil pemrosesan tersimpan di hilir” ada jendela waktu yang tidak bisa dihilangkan: consumer harus memproses pesan, lalu men commit offset. Bila consumer berhenti tepat di antara keduanya, keputusan urutan commit menentukan nasib pesan — komit duluan berarti pesan bisa terlewati, komit belakangan berarti pesan bisa diproses dua kali. Chapter ini merumuskan tiga janji pengiriman (delivery semantics), membuktikan konsekuensinya lewat simulasi crash dengan titik henti yang sama untuk ketiga mode, lalu menutup dengan pola transaksi Kafka yang dipakai pipeline produksi.

Tiga Janji Pengiriman #

Setiap pipeline streaming memilih salah satu dari tiga janji berikut — tidak ada opsi keempat.

Semantik	Janji ke hilir	Harga yang dibayar
at-most-once	Pesan boleh hilang, tidak pernah dobel	Tidak ada kerja ulang
at-least-once	Pesan tidak hilang, boleh dobel	Hilir wajib menangani duplikat
exactly-once	Tepat satu kali, apa pun yang terjadi	Transaksi, overhead, cakupan terbatas

Janji lahir dari satu keputusan teknis: kapan offset di-commit relatif terhadap pemrosesan.

Commit sebelum proses (at-most-once): offset maju lebih dulu; crash setelahnya membuat pesan yang belum diproses terlewati selamanya.
Proses sebelum commit (at-least-once): hasil aman dulu baru offset maju; crash sebelum commit membuat batch diproses ulang setelah restart.
Proses dan commit sebagai satu kesatuan (exactly-once): hasil dan offset berkomit bersama dalam satu transaksi; crash membatalkan keduanya.

Simulasi: Crash di Titik yang Sama #

Ketiga mode di bawah membaca topic 100 pesan dengan commit per 10 pesan. Consumer berhenti mendadak setelah 42 pesan — titik henti ditentukan seeded random sehingga ketiga mode menghadapi kejadian identik — lalu menyala kembali dan menyelesaikan sisanya. “Hilir” dimodelkan list sink, tempat hasil pemrosesan terlihat.

python
"""Simulasi tiga delivery semantics dengan crash di titik yang sama."""

import random

TOTAL = 100   # jumlah pesan yang dikirim producer
BATCH = 10    # offset di-commit setiap 10 pesan

# titik crash: di tengah batch ke-5 (pesan ke-41 s.d. ke-50)
CRASH_PADA = random.Random(42).randint(41, 49)


def jalankan(mode):
    sink = []            # pesan yang terlihat di hilir (misal: tabel hasil)
    offset_terkomit = 0  # posisi baca yang tersimpan di __consumer_offsets
    diproses = 0         # total pesan yang pernah lewat tangan consumer
    crash_sudah = False
    catatan_crash = ""

    while True:
        # sesi baru (awal / setelah restart): baca dari offset terakhir di-commit
        posisi = offset_terkomit
        berhenti = False
        while posisi < TOTAL:
            akhir = min(posisi + BATCH, TOTAL)
            if mode == "at-most-once":
                offset_terkomit = akhir   # commit DULU, proses kemudian
            traksi = []                   # buffer transaksi (exactly-once)
            for pesan in range(posisi, akhir):
                if not crash_sudah and diproses == CRASH_PADA:
                    crash_sudah = True
                    berhenti = True
                    break
                diproses += 1
                if mode == "exactly-once":
                    traksi.append(pesan)  # hasil ditampung di transaksi
                else:
                    sink.append(pesan)    # langsung terlihat di hilir
            if berhenti:
                break                     # consumer berhenti; traksi terbuang
            if mode == "exactly-once":
                sink.extend(traksi)       # commit: hasil + offset jadi satu
            offset_terkomit = akhir
            posisi = akhir
        if not berhenti:
            break
        catatan_crash = (f"[{mode}] berhenti setelah {diproses} pesan, "
                         f"offset terkomit = {offset_terkomit}, restart...")

    unik = len(set(sink))
    return (mode, len(sink), unik, len(sink) - unik, TOTAL - unik,
            diproses - unik, catatan_crash)


hasil = [jalankan(m) for m in ["at-most-once", "at-least-once", "exactly-once"]]

print(f"Producer mengirim {TOTAL} pesan, commit per {BATCH} pesan")
print(f"Crash disuntik setelah {CRASH_PADA} pesan (tengah batch ke-5)\n")
for baris in hasil:
    print(baris[6])
print()
print(f"{'Mode':<14}{'Terlihat':>9}{'Unik':>6}{'Duplikat':>10}{'Hilang':>8}{'Diproses ulang':>16}")
for mode, terlihat, unik, dup, hilang, ulang, _ in hasil:
    print(f"{mode:<14}{terlihat:>9}{unik:>6}{dup:>10}{hilang:>8}{ulang:>16}")

Output yang Dihasilkan:

text
Producer mengirim 100 pesan, commit per 10 pesan
Crash disuntik setelah 42 pesan (tengah batch ke-5)

[at-most-once] berhenti setelah 42 pesan, offset terkomit = 50, restart...
[at-least-once] berhenti setelah 42 pesan, offset terkomit = 40, restart...
[exactly-once] berhenti setelah 42 pesan, offset terkomit = 40, restart...

Mode           Terlihat  Unik  Duplikat  Hilang  Diproses ulang
at-most-once         92    92         0       8               0
at-least-once       102   100         2       0               2
exactly-once        100   100         0       0               2

Mari kita telaah baris-baris kode di atas:

Baris 9 (titik crash): satu angka acak berseed menentukan kapan consumer berhenti; karena angkanya sama untuk semua mode, perbandingan hasil menjadi adil.
Baris 21 (sesi baru): setiap restart selalu membaca dari offset_terkomit — baris ini satu-satunya sumber perbedaan antar mode.
Baris 25-26 (at-most-once): offset maju ke batas batch sebelum satu pun pesan diproses, sehingga crash di tengah batch meninggalkan celah.
Baris 28-32 (injeksi crash): pemeriksaan terjadi sebelum tiap pesan; setelah 42 pesan, consumer berhenti tepat di tengah batch ke-5.
Baris 34-37 (dua jalur hasil): exactly-once menampung hasil di traksi, mode lain langsung menulis ke sink.
Baris 40-41 (commit atomik): hasil dan offset hanya berlaku bersamaan — crash di tengah membatalkan keduanya.

Angka pada output menceritakan kisah lengkap. At-most-once meninggalkan 8 pesan (43 sampai 50) yang tidak pernah diproses karena offset sudah melompat ke 50. At-least-once memproses ulang pesan 41 dan 42 sehingga hilir melihat 102 pesan dengan 2 duplikat. Exactly-once juga mengulang kerja 2 pesan di dalam consumernya, tetapi transaksi membuang hasil upaya pertama sehingga hilir menerima tepat 100.

Exactly-Once dengan Transaksi Kafka #

Simulasi kita memodelkan transaksi sebagai list traksi. Di Kafka, wadah nyatanya adalah producer dengan transactional.id — identitas stabil yang bertahan lintas restart. Producer lama yang masih hidup akan dipagari otomatis begitu producer baru memakai transactional.id yang sama, mencegah dua proses menulis bersamaan. Kode berikut diverifikasi sintaksnya dengan py_compile dan tidak dijalankan di lingkungan ini:

python
"""Pola consume-process-produce transaksional (API confluent-kafka)."""

from confluent_kafka import Consumer, Producer

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "pipeline-pesanan-1",  # identitas stabil lintas restart
})
producer.init_transactions()

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "pipeline-pesanan",
    "enable.auto.commit": "false",        # offset ikut commit transaksi
    "isolation.level": "read_committed",  # hanya baca hasil transaksi selesai
})
consumer.subscribe(["pesanan"])

while True:
    batch = consumer.consume(num_messages=50, timeout=1.0)
    if not batch:
        continue

    producer.begin_transaction()
    for pesan in batch:
        producer.produce("pesanan-olahan", value=pesan.value())
    # hasil olahan + offset yang dibaca digabung dalam SATU commit transaksi
    producer.send_offsets_to_transaction(
        consumer.position(consumer.assignment()),
        consumer.consumer_group_metadata(),
    )
    producer.commit_transaction()

init_transactions() mendaftarkan identitas transaksional, begin_transaction() membuka sesi, dan commit_transaction() menuliskan hasil olahan beserta offset konsumsi sebagai satu kesatuan. Consumer di hilir memakai isolation.level=read_committed sehingga hanya melihat batch yang transaksinya selesai — persis perilaku sink.extend(traksi) pada simulasi. Rincian konfigurasi transaksi terdokumentasi di dokumentasi Apache Kafka .

Memilih Semantik untuk Pipeline #

Kebanyakan pipeline data produksi berjalan dengan at-least-once dan hilir yang idempotent: duplikat kecil lebih mudah didedup daripada pesanan yang hilang diam-diam. At-most-once layak untuk data yang boleh dibuang — metrik internal, log heartbeat — karena keterlambatan satu pesan tidak mengubah keputusan. Exactly-once dipakai ketika jalurnya memang Kafka-ke-Kafka dan biaya transaksinya terbayar oleh kebutuhan akurasi, misalnya agregasi pembayaran.

Satu dimensi ketepatan masih tersisa setelah janji pengiriman terpilih: kebenaran waktu. Pesanan pukul 23:59 yang tiba pukul 00:01 tetap terkirim tepat satu kali, tetapi masuk laporan tanggal yang salah — masalah ini bukan soal pengiriman, melainkan soal dua jam yang berbeda, dan menjadi pokok bahasan Chapter Event Time vs Processing Time.