Eksperimen Bagian 6 berakhir dengan fakta yang tidak nyaman: mode at-least-once yang kita pilih mengirim ulang lima event setelah consumer pulih, dan hilir menerima evt-000021 sampai evt-000025 dua kali. Untuk agregasi jumlah pesanan, duplikat berarti angka jendela menggelembung; untuk sistem notifikasi, pelanggan menerima dua email. Solusinya adalah consumer idempoten: hilir mengingat event_id yang sudah diproses, memakai ingatan itu untuk menyaring pengiriman ulang, dan menjaga urutan tiga langkah — proses, simpan state, commit offset — agar tidak ada celah duplikat maupun kehilangan. Bagian ini membangun mekanisme tersebut dan menguji dua titik henti yang berbeda. Pola yang sama pada dunia batch sudah kita pakai di chapter Idempotent Ingestion; di sini ia bekerja di atas Kafka.

State Hilir yang Tahan Crash #

Kunci idempotensi ada pada HasilHilir di src/idempotent.py: agregat jendela dan jurnal event_id hidup dalam satu kesatuan yang dimuat bersama saat consumer menyala dan disimpan bersama secara atomik:

python 
    def __init__(self, path: str | Path | None = None) -> None:
        self._path = Path(path) if path is not None else None
        self.jurnal: set[str] = set()
        self._agregator = AgregatorJendela()
        if self._path is not None and self._path.exists():
            isi = json.loads(self._path.read_text(encoding="utf-8"))
            self.jurnal = set(isi["diproses"])
            for kunci, kotak in isi["jendela"].items():
                self._agregator.jendela[kunci] = dict(kotak)

    def sudah_diproses(self, event_id: str) -> bool:
        """Apakah ``event_id`` pernah diproses sebelumnya?"""
        return event_id in self.jurnal

    def tambah_event(self, event: dict) -> None:
        """Agregasikan satu event lalu catat ``event_id``-nya ke jurnal."""
        self._agregator.tambah_event(event)
        self.jurnal.add(event["event_id"])

    def simpan(self) -> None:
        """Tulis jurnal + jendela ke disk secara atomik (tmp + os.replace)."""
        if self._path is None:
            return
        self._path.parent.mkdir(parents=True, exist_ok=True)
        isi = {
            "diproses": sorted(self.jurnal),
            "jendela": self._agregator.ringkasan(),
        }
        sementara = self._path.with_suffix(".tmp")
        sementara.write_text(json.dumps(isi, indent=1), encoding="utf-8")
        os.replace(sementara, self._path)

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

  1. Baris 29-37 (__init__): saat consumer baru dibuat, jurnal dan jendela dimuat dari satu berkas yang sama — tidak ada jendela yang dibangun ulang tanpa jurnalnya, dan sebaliknya.
  2. Baris 39-41 (sudah_diproses): pemeriksaan duplikat yang dipanggil proses_records untuk setiap event sebelum diproses.
  3. Baris 43-46 (tambah_event): agregasi jendela dan pencatatan event_id dilakukan dalam satu panggilan — dua perubahan ini harus terjadi bersama atau tidak sama sekali.
  4. Baris 48-59 (simpan): penulisan lewat berkas sementara lalu os.replace (baris 59) bersifat atomik di level sistem berkas: pembaca lain melihat isi lama yang utuh atau isi baru yang utuh, tidak pernah setengah jadi.

Menyaring Duplikat #

Saringan duplikat berada di ConsumerIdempoten.proses_records — event yang event_id-nya sudah tercatat dilewati, event baru diagregasi lalu dicatat:

python 
    def proses_records(self, records: list) -> dict[str, int]:
        """Proses satu batch; kembalikan jumlah event baru dan duplikat."""
        baru = 0
        duplikat = 0
        for rekam in records:
            event = deserialisasi_event(rekam)
            if self._hasil.sudah_diproses(event["event_id"]):
                duplikat += 1
                continue
            self._hasil.tambah_event(event)
            baru += 1
        return {"baru": baru, "duplikat": duplikat}

Urutan tiga langkah per batch inilah jaminan kita: proses_records (proses), consumer.simpan() (simpan state hilir), konsumen.commit() (maju-kan offset). Commit selalu terakhir; selama urutan itu berlaku, pesan yang belum terkomit akan terkirim ulang setelah consumer pulih — dan jurnal menyaringnya.

Dua Titik Henti, Satu Hasil Akhir #

Fungsi jalankan_dengan_kegagalan menyuntikkan henti pada dua titik berbeda. Bagian intinya:

python 
    while True:
        # Setiap restart memuat ulang state hilir dari disk; hasil milik
        # consumer yang berhenti sebelum sempat disimpan ikut terbuang.
        hasil = HasilHilir(state_path)
        consumer = ConsumerIdempoten(hasil)
        konsumen = buat_consumer(
            broker=broker,
            group_id=group_id,
            enable_auto_commit=False,
            auto_offset_reset="earliest",
            max_poll_records=ukuran_batch,
        )
        konsumen.subscribe([topic])

Titik henti pertama (crash_mode="sebelum_komit"): batch ke-2 diproses dan jurnal tersimpan, lalu consumer berhenti tepat sebelum commit — batch yang sama terkirim ulang setelah pulih dan jurnal menyaring semuanya. Titik kedua (crash_mode="saat_proses"): consumer berhenti setelah event ke-25 di tengah batch ketiga, sebelum simpan() — hasil sebagian batch ikut hangus bersama state di memori, dan seluruh batch diproses ulang. Perintah python3 -m src.idempotent menjalankan keduanya:

Output yang Dihasilkan:

text 
crash: berhenti setelah jurnal tersimpan, sebelum commit
  dikirim=60 pemrosesan=70 duplikat_disisihkan=10 hilang=0 event_unik=60
crash: berhenti setelah event ke-25, di tengah batch
  dikirim=60 pemrosesan=65 duplikat_disisihkan=0 hilang=0 event_unik=60

Bacaan hasilnya:

Titik hentiPemrosesanDuplikat disaringHasil akhir
setelah simpan, sebelum commit70 (batch ke-2 terkirim ulang utuh)1060 event, tepat satu kali
di tengah batch, sebelum simpan65 (5 event hangus + diproses ulang)060 event, tepat satu kali

Pada skenario pertama, sepuluh event dikirim ulang dan jurnal menyaring semuanya — pemrosesan=70 mencakup sepuluh kunjungan ulang yang tidak mengubah hasil. Pada skenario kedua, lima event terproses di memori lalu hangus bersama consumer; state di disk tidak pernah mencatatnya, sehingga setelah pulih keduanya diproses ulang sebagai event baru — jurnal tidak menyaring apa pun (duplikat_disisihkan=0) karena tidak ada yang sempat tercatat. Kedua jalur berbeda, ujungnya sama: event_unik=60, hilang=0, dan jendela hilir berisi tepat 60 event.

Memeriksa State Hilir #

State yang tersimpan di disk bisa diperiksa langsung setelah skenario pertama (kutipan demo_state_hilir.py):

python 
    isi = json.loads(state.read_text(encoding="utf-8"))
    print(f"jurnal berisi {len(isi['diproses'])} event_id, "
          f"5 pertama: {isi['diproses'][:5]}")
    print("jendela hasil hilir (3 pertama):")
    for kunci in list(isi["jendela"])[:3]:
        print(f"  {kunci}: {isi['jendela'][kunci]}")

Output yang Dihasilkan:

text 
dikirim=60 pemrosesan=70 duplikat_disisihkan=10 hilang=0 event_unik=60
jurnal berisi 60 event_id, 5 pertama: ['evt-000001', 'evt-000002', 'evt-000003', 'evt-000004', 'evt-000005']
jendela hasil hilir (3 pertama):
  2024-03-01T09:00: {'jumlah': 5, 'total': 7258500}
  2024-03-01T09:01: {'jumlah': 5, 'total': 4670500}
  2024-03-01T09:02: {'jumlah': 5, 'total': 6234500}

Jurnal berisi tepat 60 event_id — bukan 70 — dan setiap jendela menampung lima event, bukan lima ditambah salinan ulang. Inilah efek proses tepat satu kali (effectively exactly-once) yang dibangun di atas at-least-once: Kafka masih mengirim ulang, tetapi hilir sudah kebal. Seluruh skenario ini dikunci oleh tes test_jalankan_crash_sebelum_komit dan test_jalankan_crash_saat_proses.

Consumer kita kini tahan gagal, tetapi semua perlindungan itu runtuh bila pesan yang masuk sendiri tidak layak diproses — event dari producer lain yang berubah bentuk, field hilang, atau versi schema baru. Bagian berikutnya memasang gerbang terakhir sebelum hilir: validasi event.