Producer menulis; sekarang kita baca. Consumer Kafka tidak menerima push dari broker — consumer memanggil poll() secara berulang, memproses batch yang kembali, lalu mengommit posisi bacanya. Pola ini memberi kendali penuh atas satu keputusan penting: kapan offset dinyatakan selesai, sebelum atau sesudah hasil pemrosesan tersimpan. Bagian ini membangun consumer group agregator-pesanan dengan commit manual setelah hasil aman (pola at-least-once), lalu mengagregasikan event pesanan menjadi jendela per menit — jumlah event dan total nilai — sebagai produk hilir stream TokoKita. Peran consumer group dan pembagian partition sudah dibedah pada chapter Consumer Groups.

Poll Loop dengan Commit Manual #

Fungsi jalankan_agregator di src/consumer.py memuat seluruh pola konsumsi proyek ini:

python 
def jalankan_agregator(
    broker,
    topic: str,
    group_id: str = "agregator-pesanan",
    ukuran_batch: int = 100,
) -> dict:
    """Jalankan poll loop sampai topic habis; commit per batch (at-least-once).

    Mengembalikan jumlah event terproses, ringkasan jendela, dan posisi
    committed offset akhir tiap partition.
    """
    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])
    agregator = AgregatorJendela()
    jumlah_event = 0
    while True:
        batch = konsumen.poll()
        if not batch:
            break
        records = [rekam for daftar in batch.values() for rekam in daftar]
        agregator.proses_records(records)
        jumlah_event += len(records)
        konsumen.commit()
    offset_komit = {
        tp: konsumen.committed(tp) for tp in sorted(konsumen.assignment())
    }
    konsumen.close()
    return {
        "event": jumlah_event,
        "jendela": agregator.ringkasan(),
        "offset_komit": offset_komit,
    }

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

  1. Baris 82-87 (buat_consumer(...)): consumer dibuat dengan enable_auto_commit=False — komit menjadi tanggung jawab kode kita — dan auto_offset_reset="earliest" sehingga grup yang belum punya committed offset mulai membaca dari offset 0. Pabrik buat_consumer mengikuti protokol yang sama dengan producer: backend="local" untuk simulasi, backend="kafka" untuk cluster nyata.
  2. Baris 89 (subscribe): bergabung ke consumer group agregator-pesanan. Group inilah unit rebalance: bila dua consumer memakai group yang sama, tiga partition topic dibagi dua dan tidak ada pesan yang diproses ganda di dalam satu group.
  3. Baris 93-96 (while True / poll()): loop inti consumer — poll() mengembalikan pemetaan TopicPartition ke daftar record sampai batas max_poll_records; batch kosong berarti seluruh histori terkejar.
  4. Baris 97-99 (proses lalu komit): urutan ini adalah jaminan at-least-once — hasil agregasi masuk dulu ke agregator, baru offset dikomit. Bila consumer berhenti di antara keduanya, batch dikirim ulang setelah pulih; konsekuensinya kita tangani pada Bagian 7.
  5. Baris 100-103 (offset_komit): posisi committed offset tiap partition dicatat sebagai bukti akhir — angka ini yang dibandingkan dengan end_offsets untuk menghitung lag pada Bagian 9.

Jendela Per Menit dari Event Time #

Agregasi membagi stream menjadi jendela menit berdasarkan event_time — bukan waktu pemrosesan — sehingga hasil tidak bergeser walau consumer telat (perbedaan keduanya dibedah pada chapter Event Time vs Processing Time dan Windowing):

python 
def jendela_menit(event_time: str) -> str:
    """Kunci jendela per menit dari stempel waktu ISO 8601 event.

    Contoh: ``"2024-03-01T09:07:34+00:00"`` menjadi ``"2024-03-01T09:07"``.
    """
    return event_time[:16]

Kunci jendela berupa potongan 16 karakter ISO 8601 (YYYY-MM-DDTHH:MM), dan AgregatorJendela.tambah_event menampung jumlah serta total nilai per kunci:

python 
    def tambah_event(self, event: dict) -> str:
        """Tambahkan satu event ke jendelanya; kembalikan kunci jendela."""
        kunci = jendela_menit(event["event_time"])
        kotak = self.jendela.setdefault(kunci, {"jumlah": 0, "total": 0})
        kotak["jumlah"] += 1
        kotak["total"] += event["total"]
        return kunci

Deserialisasi nilai pesan memakai modul json stdlib — deserialisasi_event mengurai record.value (bytes) menjadi dict event yang sama seperti yang dikirim producer.

Menjalankan Consumer #

Simulasi berikut mengirim 30 event lalu menjalankan agregator penuh:

python 
from src.local_kafka import LocalKafka
from src.producer import SimulatorCheckout, buat_producer
from src.consumer import jalankan_agregator

broker = LocalKafka()
broker.create_topic("tokokita.pesanan", num_partitions=3)
producer = buat_producer(broker=broker, client_id="checkout-web")
SimulatorCheckout(producer, topic="tokokita.pesanan").kirim(30)

hasil = jalankan_agregator(broker, "tokokita.pesanan", group_id="agregator-pesanan")
print(f"event terproses : {hasil['event']}")
print(f"jumlah jendela  : {len(hasil['jendela'])}")
print("jendela per menit (jumlah, total nilai):")
for kunci, isi in hasil["jendela"].items():
    print(f"  {kunci}: {isi['jumlah']} event, total {isi['total']}")
print("committed offset per partition:")
for tp, komit in hasil["offset_komit"].items():
    print(f"  partition {tp.partition}: {komit}")

Output yang Dihasilkan:

text 
event terproses : 30
jumlah jendela  : 6
jendela per menit (jumlah, total nilai):
  2024-03-01T09:00: 5 event, total 7258500
  2024-03-01T09:01: 5 event, total 4670500
  2024-03-01T09:02: 5 event, total 6234500
  2024-03-01T09:03: 5 event, total 5169500
  2024-03-01T09:04: 5 event, total 5445000
  2024-03-01T09:05: 5 event, total 7529000
committed offset per partition:
  partition 0: 13
  partition 1: 5
  partition 2: 12

Perhatikan angka committed offset: partition 0 diakhiri di 13, partition 1 di 5, partition 2 di 12 — jumlahnya tepat 30, dan masing-masing sama dengan end_offsets partition tersebut. Committed offset berarti “pesan berikutnya yang akan dibaca”, sehingga consumer group yang sama tidak akan membaca ulang 30 event ini; group lain yang baru pertama kali membaca topic akan menerima histori yang sama dari offset 0 (perilaku yang diuji test_dua_group_membaca_histori_sama).

Stream kini mengalir dua arah: producer mengisi topic, consumer menghasilkan jendela per menit. Sebelum membahas kegagalan, kita perlu memahami satu hal yang menentukan beban keduanya: bagaimana kunci pesan memetakan event ke partition.