Bagian 9: Memantau Stream

Pipeline batch yang gagal terlihat besok pagi saat DAG berwarna merah; pipeline streaming yang gagal bisa diam-diam berhenti menghasilkan sementara event terus menumpuk tanpa diproses. Pertanyaan pemantauan streaming karenanya berbeda: bukan “apakah run berhasil”, melainkan “seberapa jauh consumer tertinggal sekarang dan berapa laju event yang ia tangani”. Bagian penutup ini membangun pemantauan dari dua sumber yang sudah kita miliki — offset akhir partition dan committed offset group — menghasilkan laporan lag dan metrik jendela dari eksekusi nyata, lalu memetakan jembatan ke ekosistem pemantauan Kafka sungguhan. Kerangka metrik dan alerting-nya mengikuti chapter Monitoring Streaming Pipeline dan Metrics dan Alerting.

Consumer Lag: Jarak terhadap Ekor #

Consumer lag adalah metrik kesehatan terpenting stream: selisih offset akhir partition (log terus bertambah) dengan committed offset group (posisi baca terakhir). Fungsi hitung_lag di src/monitor.py menghitungnya langsung dari state broker:

def hitung_lag(broker, group_id: str, topic: str) -> dict[TopicPartition, int]:
    """Consumer lag tiap partition: offset akhir dikurangi committed offset.

    Group yang belum pernah mengommit dihitung tertinggal sebesar seluruh
    isi partition, setara consumer baru yang belum membaca apa pun.
    """
    tps = [TopicPartition(topic, p) for p in broker.partitions_for_topic(topic)]
    akhir = broker.end_offsets(tps)
    lag = {}
    for tp in tps:
        komit = broker.committed(group_id, tp)
        lag[tp] = akhir[tp] - (komit if komit is not None else 0)
    return lag

Interpretasinya sederhana namun kaya makna: lag 0 berarti consumer menempel di ekor log; lag yang tumbuh terus berarti laju produksi melampaui laju konsumsi (consumer terlalu lambat, partition panas, atau consumer berhenti); lag besar yang tiba-tiba turun ke 0 bisa berarti replay sedang berjalan. Perhatikan pula bahwa lag dihitung per partition — partition 1 yang memikul 81 pesan pada demo Bagian 5 layak dipantau terpisah dari yang lain.

Metrik per Jendela #

Lag menjawab “sehat tidaknya”; metrik jendela menjawab “seberapa cepat dan berapa nilai yang mengalir”. metrik_jendela dan throughput_total menghitungnya dari hasil agregator:

def throughput_total(jendela: dict) -> dict:
    """Throughput keseluruhan: total event dibagi durasi jendela yang tercakup."""
    if not jendela:
        return {"event": 0, "durasi_detik": 0.0, "event_per_detik": 0.0}
    kunci = sorted(jendela)
    awal = datetime.fromisoformat(kunci[0] + ":00+00:00")
    akhir = datetime.fromisoformat(kunci[-1] + ":00+00:00")
    durasi = (akhir - awal).total_seconds() + 60
    event = sum(isi["jumlah"] for isi in jendela.values())
    return {
        "event": event,
        "durasi_detik": durasi,
        "event_per_detik": round(event / durasi, 3),
    }

Durasi dihitung dari rentang jendela pertama sampai terakhir ditambah satu menit penuh — bukan dari jam mesin — sehingga angkanya stabil lintas eksekusi. Fungsi metrik_jendela menambahkan per_detik (jumlah event dibagi 60 detik) dan rata_rata nilai pesanan per jendela.

Laporan Lengkap #

Perintah python3 -m src.monitor menjalankan alur penuh: mengisi topic dengan 120 event berkunci wilayah, mengukur lag sebelum konsumsi, menjalankan agregator, lalu mencetak laporan:

Output yang Dihasilkan (bagian awal dan akhir; 21 baris jendela di tengah dipangkas):

statistik topic sebelum dikonsumsi:
  partition 0: 31 pesan (offset akhir 30)
  partition 1: 81 pesan (offset akhir 80)
  partition 2: 8 pesan (offset akhir 7)
lag grup agregator-pesanan sebelum konsumsi: 120
event terproses: 120

=== Laporan Stream: tokokita.pesanan ===
consumer group : agregator-pesanan
total event    : 120 (0.083 event/detik selama 1440 detik)

jendela            jumlah  event/detik  total nilai   rata-rata
2024-03-01T09:00      5        0.083      7258500   1451700.0
...
2024-03-01T09:23      5        0.083      6739000   1347800.0

consumer lag:
  partition 0: lag 0
  partition 1: lag 0
  partition 2: lag 0
  total lag: 0

Angka-angka itu saling menjelaskan. Lag 120 sebelum konsumsi persis sama dengan total pesan di topic — group belum membaca apa pun. Setelah agregator selesai, lag tiga partition nol dan jendela 09:00 sampai 09:23 menampung 5 event per menit, konsisten dengan jarak event 12 detik dari Bagian 1. Skew Bagian 5 juga terlihat: partition 1 menampung 81 pesan, dua pertiga dari seluruh stream.

Jembatan ke Ekosistem Nyata #

Pemantauan di atas bekerja pada simulasi satu proses; cluster produksi menuntut alat yang berjalan terus-menerus dan bisa memberi peringatan. Padanan langsungnya di ekosistem Kafka:

Alur produksi yang lazim: Kafka Exporter (atau Burrow) membaca offset akhir dan committed offset dari cluster lalu mengekspos lag sebagai metrik; Prometheus mengambil metrik itu secara berkala; Grafana menampilkannya; aturan alert memberi peringatan ketika lag tumbuh melewati ambang selama beberapa menit. Tidak ada satu pun angka baru yang muncul di rantai itu — semuanya berasal dari pasangan end_offsets dan committed offset yang sama dengan hitung_lag; yang berubah hanya cara pengumpulannya yang terus-menerus. Detail operasionalnya dibahas pada chapter Monitoring Streaming Pipeline dan Metrics dan Alerting.

Rekap Sembilan Bagian #

Sebelum menutup, mari kita rapikan apa yang sudah dibangun dan berkas yang menaunginya:

Dua keputusan arsitektur menaungi semuanya: pertama, seluruh kode ditulis menghadap protokol kecil (send/flush untuk producer, subscribe/poll/commit untuk consumer) sehingga simulasi dan cluster nyata bisa dipertukarkan lewat satu argumen backend; kedua, setiap klaim perilaku — dari vektor hash murmur2 sampai efek tepat-satu-kali setelah crash — dikunci oleh tes, bukan oleh asumsi.

Pengujian Akhir #

Seluruh perilaku yang dibangun selama sembilan bagian dikunci oleh suite pytest proyek:

python3 -m pytest tests/ -q

Output yang Dihasilkan:

........................................................................ [ 85%]
............                                                             [100%]
84 passed in 0.31s

Delapan puluh empat tes mencakup sembilan berkas: semantik broker (offset monoton, rebalance, seek), partitioner dengan vektor resmi Kafka, producer dan consumer, eksperimen commit offset dengan crash, consumer idempoten dengan dua titik henti, validator schema, monitor, dan alur end-to-end dari produksi sampai laporan. Karena semuanya berjalan di atas LocalKafka dalam satu proses tanpa I/O jaringan, suite selesai di bawah satu detik — sifat yang sama yang membuat simulasi ini cocok untuk latihan; untuk cluster nyata, Anda tinggal mengganti backend="kafka" dan menjalankan ulang tes integrasi terhadap cluster KRaft dari Bagian 2. Konvensi pengujian pytest yang dipakai terdokumentasi di dokumentasi pytest .

Proyek ini menutup perjalanan streaming kita: event dirancang sebagai fakta, dialirkan lewat topic multi-partition dengan kunci terukur, dikonsumsi dengan jaminan at-least-once yang diberi idempotensi di hilir, dijaga schema-nya, dan dipantau lag-nya. Langkah alami berikutnya adalah memproses stream dalam skala besar — transformasi, windowing, dan join lintas stream dengan mesin pemrosesan terdistribusi — yang akan kita kerjakan pada Proyek: Pemrosesan Data dengan Spark.