Bagian 6: Mengelola Offset

Committed offset adalah satu-satunya memori consumer group: angka per partition yang menyatakan “pesan berikutnya yang akan dibaca”. Kapan angka itu maju — sebelum atau sesudah hasil pemrosesan tersimpan — menentukan nasib pesan saat consumer berhenti mendadak: komit duluan berarti pesan bisa terlewati, komit belakangan berarti pesan bisa diproses dua kali. Chapter Delivery Semantics merumuskan tiga janji pengiriman; bagian ini membuktikan dua di antaranya dengan eksperimen terkendali: 60 event, batch 10, dan consumer yang berhenti tepat setelah event ke-25 — titik henti yang sama untuk kedua mode. Ditutup dengan replay: memutar ulang histori dari offset 0 tanpa menyentuh committed offset group.

Posisi Baca Dimuat dari Committed Offset #

Ketika consumer group baru memulai — atau partition berpindah pemilik setelah rebalance — posisi baca ditentukan oleh metode internal _muat_posisi pada src/local_kafka.py:

python
    def _muat_posisi(self, tp: TopicPartition) -> None:
        komit = self._broker.committed(self.group_id, tp)
        if komit is not None:
            self._posisi[tp] = komit
        elif self.auto_offset_reset == "earliest":
            self._posisi[tp] = 0
        elif self.auto_offset_reset == "latest":
            self._posisi[tp] = self._broker.end_offsets([tp])[tp]
        else:
            raise NoOffsetForPartitionError(
                f"{tp}: tidak ada committed offset dan auto_offset_reset nonaktif"
            )

Aturannya berjenjang: committed offset menang bila ada; tanpa komit, earliest memulai dari offset 0 (membaca histori), latest melompat ke ekor (hanya pesan baru). Karena itu consumer group yang crash di tengah jalan otomatis melanjutkan dari komit terakhirnya saat menyala kembali — mekanisme inilah yang membuat kedua eksperimen berikut bisa dijalankan tanpa menyimpan state apa pun di sisi aplikasi.

Dua Urutan Commit, Satu Titik Henti #

Fungsi jalankan_semantik di src/offsets.py menjalankan satu mode dengan injeksi henti pada event ke-crash_setelah:

python
        while True:
            batch = konsumen.poll()
            if not batch:
                break
            records = [r for daftar in batch.values() for r in daftar]
            if semantik == AT_MOST_ONCE:
                konsumen.commit()  # offset maju lebih dulu
                for rekam in records:
                    proses_satu(rekam)
                    if state["crash"]:
                        crash_di_run_ini = True
                        break
            else:
                for rekam in records:
                    proses_satu(rekam)
                    if state["crash"]:
                        crash_di_run_ini = True
                        break
                if not crash_di_run_ini:
                    konsumen.commit()  # hasil aman dulu, offset menyusul
            if crash_di_run_ini:
                break
        konsumen.close()
        if not crash_di_run_ini:
            break  # poll kosong: seluruh histori selesai
        state["crash"] = False  # consumer baru menyala kembali

Mari kita telaah baris-baris kode di atas (nomor dihitung dari atas blok, karena kutipan mengambil bagian tengah fungsi):

Baris 7 (konsumen.commit() di cabang pertama): mode at-most-once mengommit seluruh batch sebelum memproses satu pun event — offset sudah maju ke 30 ketika batch ketiga (event 21-30) diambil.
Baris 8-12 (proses dengan cek henti): event diproses satu per satu; setelah event ke-25, state["crash"] aktif dan loop putus — lima event batch ketiga (26-30) sudah terkomit tetapi tidak pernah diproses.
Baris 13-20 (cabang at-least-once): urutan dibalik — proses dulu, komit hanya bila batch selesai. Henti di event ke-25 membatalkan komit batch ketiga, sehingga committed offset tetap 20.
Baris 23-26 (restart): consumer ditutup lalu loop luar membuat consumer baru dengan group yang sama; _muat_posisi mengembalikan posisi dari committed offset, dan consumer baru melanjutkan — membaca ulang event 21-30 pada mode at-least-once, atau melompati 26-30 permanen pada mode at-most-once.

Hasil Eksperimen #

Perintah python3 -m src.offsets menjalankan keduanya pada topic yang sama dengan group terpisah:

Output yang Dihasilkan:

text
Dikirim 60 event; consumer berhenti tepat setelah event ke-25 (batch 10 event)
at-most-once   pemrosesan= 55 unik= 55 duplikat= 0 hilang= 5
at-least-once  pemrosesan= 65 unik= 60 duplikat= 5 hilang= 0
replay dari offset 0: 60 event dibaca ulang

Kedua mode menghadapi titik henti yang sama namun menderita di tempat yang berlawanan. At-most-once kehilangan lima event secara permanen — offset sudah maju ke 30 ketika consumer berhenti di event ke-25, sehingga event 26-30 terlewati selamanya meskipun consumer pulih. At-least-once tidak kehilangan apa pun, tetapi membayar dengan lima pemrosesan ulang: committed offset tertinggal di 20, dan setelah pulih consumer membaca ulang seluruh batch ketiga — event 21-25 diproses dua kali. Rangkumannya:

Mode	Urutan	5 event sisa batch ke-3	Konsekuensi
at-most-once	commit → proses	terkomit, tidak diproses	hilang permanen (5 event)
at-least-once	proses → commit	tidak terkomit, dikirim ulang	diproses dua kali (5 event)

Replay dari Offset #

Karena log partition bersifat append-only dan retensi belum membersihkan apa pun, histori selalu bisa dibaca ulang. Fungsi replay_dari_offset memakai seek() — menggeser posisi baca satu partition tanpa mengommit:

python
    konsumen.subscribe([topic])
    for tp in konsumen.assignment():
        konsumen.seek(tp, offset)
    records = []
    while True:
        batch = konsumen.poll()
        if not batch:
            break
        records.extend(r for daftar in batch.values() for r in daftar)
    konsumen.close()
    return records

Demo berikut memutar ulang 60 event setelah kedua eksperimen selesai (kutipan demo_replay.py):

python
records = replay_dari_offset(broker, "tokokita.pesanan", "audit-at-least-once")
print(f"replay dari offset 0: {len(records)} event dibaca ulang")

Output yang Dihasilkan (bagian pemeriksaan committed offset):

text
topic berisi 60 event
replay dari offset 0: 60 event dibaca ulang
event pertama : offset=0 partition=0
event terakhir: offset=23 partition=2
committed offset group audit-at-least-once (tidak berubah oleh replay):
  partition 0: committed=18 end=18
  partition 1: committed=18 end=18
  partition 2: committed=24 end=24

Enam puluh event terbaca ulang dari offset 0, sedangkan committed offset group tetap 18/18/24 — replay tidak meninggalkan jejak pada posisi group reguler karena tidak ada komit yang dilakukan. Pada cluster nyata, replay sekelompok consumer biasa dilakukan lewat kafka-consumer-groups.sh --reset-offsets --to-earliest --execute sebelum grup dinyalakan, atau seek() untuk kasus khusus seperti backfill satu partition; keduanya memanfaatkan fakta yang sama bahwa pesan tidak dihapus setelah dibaca.

Kita telah melihat dua dari tiga janji pengiriman dibuktikan. Tepat di titik ini masalah berikutnya muncul: pada mode at-least-once yang kita pilih, lima event terkirim ulang — sekarang saatnya membuat hilir kebal terhadap pengiriman ulang tersebut.