Menangani Kegagalan Ingestion

Semua pertahanan yang kita bangun tetap menyisakan kenyataan: pipeline akan gagal di tengah jalan. Koneksi API putus pada pesanan keempat dari enam ribu, disk penuh pukul tiga pagi, atau proses berhenti karena server direstart saat deploy. Pipeline yang buruk berhenti dan menunggu dijalankan ulang dari awal; pipeline yang baik mencatat kegagalan, berhenti di posisi aman, lalu melanjutkan dari titik tersebut. Chapter penutup section ini merangkai klasifikasi kegagalan, retry yang aman, checkpoint, dead letter queue, dan observability menjadi satu contoh utuh yang diuji dua tahap.

Klasifikasi Kegagalan: Transien versus Permanen #

Semua kegagalan jatuh ke dua kelompok dengan penanganan yang berlawanan. Kegagalan transien akan pulih dengan sendirinya jika dicoba lagi — koneksi terputus sejenak, status 429, timeout. Kegagalan permanen tidak akan pernah berhasil berapa kali pun dicoba — kredensial salah, schema sumber berubah total, atau bug di kode pipeline kita sendiri.

Menerapkan retry pada kegagalan permanen adalah pemborosan klasik: pipeline mencoba ulang sepuluh kali untuk permintaan yang ditolak karena token kedaluwarsa, lalu tetap gagal. Alat ukurnya sederhana: tanyakan apakah menunggu membuat masalah hilang; jika ya, kegagalan bersifat transien.

Retry yang Aman dan Checkpoint #

Retry hanya aman jika penulisan di tujuan bersifat idempotent, karena baris yang sudah tertulis bisa tertulis lagi setelah proses dijalankan ulang. INSERT OR IGNORE dengan dedup key dari Chapter Idempotent Ingestion menjadikan pemrosesan ulang tidak berbahaya.

Checkpoint melangkah lebih jauh: pipeline mencatat posisi terakhir yang berhasil diproses, lalu melanjutkan dari posisi tersebut. Checkpoint bisa berupa nomor baris, updated_at terakhir (seperti pada Chapter Incremental Ingestion), atau offset topic Kafka. Satu aturan urutan wajib dipegang: data harus di-commit lebih dulu, baru checkpoint diperbarui. Jika terbalik dan proses berhenti di antara keduanya, checkpoint mengklaim baris yang belum tersimpan, dan baris tersebut terlewat selamanya. Dengan urutan yang benar, kegagalan di antara keduanya hanya menyebabkan satu baris diproses ulang — aman karena penulisannya idempotent.

Dead Letter Queue dan Observability Dasar #

Pada pipeline streaming, satu baris macet tidak boleh menghentikan seluruh aliran. Pola dead letter queue (DLQ) mengarahkan baris yang gagal setelah semua retry habis ke antrean terpisah — di Kafka biasanya berupa topic khusus bernama dead letter — sementara pipeline melanjutkan baris berikutnya. Tim kemudian memeriksa isi DLQ, memperbaiki akar masalah, dan memproses ulang. Dokumentasi pola ini di ekosistem Kafka bisa dibaca di https://kafka.apache.org/documentation/ .

Observability dasar menjawab tiga pertanyaan: apa yang gagal, di mana posisinya, dan mengapa. Untuk itu, setiap kegagalan dicatat ke log dengan tiga informasi tersebut. Log kegagalan adalah sinyal termurah — satu baris per kejadian — yang cukup untuk membangun alarm di sistem monitoring.

Contoh: Ingestion dengan Checkpoint yang Di-resume #

Kita rangkai semuanya. Skrip berikut mengambil enam pesanan satu per satu, menulis ke SQLite berkas (bukan :memory:), memperbarui checkpoint setelah setiap baris ter-commit, dan mencatat kegagalan ke catat-galat.log. Kegagalan disimulasikan lewat berkas penanda gangguan-aktif: selama berkas tersebut ada, pengambilan pesanan ORD-004 selalu gagal dengan ConnectionError setelah tiga percobaan.

import json
import os
import sqlite3
import sys
import time

DIREKTORI = "keadaan_toko"
BERKAS_CHECKPOINT = os.path.join(DIREKTORI, "checkpoint.json")
BERKAS_GALAT = os.path.join(DIREKTORI, "catat-galat.log")
BERKAS_GANGGUAN = os.path.join(DIREKTORI, "gangguan-aktif")

PESANAN = [
    {"order_id": "ORD-001", "pelanggan": "Budi", "total": 150000},
    {"order_id": "ORD-002", "pelanggan": "Siti", "total": 275000},
    {"order_id": "ORD-003", "pelanggan": "Dewi", "total": 99000},
    {"order_id": "ORD-004", "pelanggan": "Budi", "total": 430000},
    {"order_id": "ORD-005", "pelanggan": "Siti", "total": 320000},
    {"order_id": "ORD-006", "pelanggan": "Dewi", "total": 87000},
]


def gangguan_aktif():
    return os.path.exists(BERKAS_GANGGUAN)


def ambil_pesanan(baris, maks_percobaan=3):
    for percobaan in range(1, maks_percobaan + 1):
        if baris["order_id"] == "ORD-004" and gangguan_aktif():
            print(f"[ingesti] ORD-004 gagal (percobaan {percobaan}): koneksi ke API terputus")
            time.sleep(0.2)
            continue
        return baris
    raise ConnectionError("koneksi ke API terputus pada ORD-004")


def muat_checkpoint():
    if os.path.exists(BERKAS_CHECKPOINT):
        with open(BERKAS_CHECKPOINT, encoding="utf-8") as f:
            return json.load(f)["terproses"]
    return 0


def simpan_checkpoint(jumlah):
    with open(BERKAS_CHECKPOINT, "w", encoding="utf-8") as f:
        json.dump({"terproses": jumlah}, f)


def main():
    os.makedirs(DIREKTORI, exist_ok=True)
    mulai = muat_checkpoint()
    if mulai > 0:
        print(f"[ingesti] mulai dari baris ke-{mulai + 1} (lanjut dari checkpoint)")
    else:
        print("[ingesti] mulai dari baris ke-1")

    db = sqlite3.connect(os.path.join(DIREKTORI, "toko.db"))
    db.execute("""CREATE TABLE IF NOT EXISTS pesanan (
        order_id TEXT PRIMARY KEY,
        pelanggan TEXT NOT NULL,
        total INTEGER NOT NULL)""")

    tersimpan = 0
    try:
        for indeks in range(mulai, len(PESANAN)):
            baris = ambil_pesanan(PESANAN[indeks])
            db.execute(
                "INSERT OR IGNORE INTO pesanan (order_id, pelanggan, total) VALUES (?, ?, ?)",
                (baris["order_id"], baris["pelanggan"], baris["total"]),
            )
            db.commit()
            simpan_checkpoint(indeks + 1)
            print(f"[ingesti] {baris['order_id']} tersimpan")
            tersimpan += 1
    except ConnectionError as galat:
        posisi = muat_checkpoint()
        with open(BERKAS_GALAT, "a", encoding="utf-8") as f:
            f.write(f"[gagal] {galat}; checkpoint tetap di baris {posisi}\n")
        print(f"[ingesti] berhenti: {galat}")
        print(f"[ingesti] checkpoint tetap di baris {posisi}; kegagalan dicatat ke catat-galat.log")
        return 1

    total = db.execute("SELECT COUNT(*) FROM pesanan").fetchone()[0]
    print(f"[ingesti] selesai: {tersimpan} baris baru, total {total} baris di tabel pesanan")
    return 0


sys.exit(main())

Tahap pertama: aktifkan berkas gangguan lalu jalankan skrip. Tiga pesanan pertama tersimpan, ORD-004 gagal tiga kali, dan skrip berhenti dengan exit code 1.

mkdir -p keadaan_toko
touch keadaan_toko/gangguan-aktif
python3 ingestion.py
echo "exit code: $?"

Output yang Dihasilkan:

[ingesti] mulai dari baris ke-1
[ingesti] ORD-001 tersimpan
[ingesti] ORD-002 tersimpan
[ingesti] ORD-003 tersimpan
[ingesti] ORD-004 gagal (percobaan 1): koneksi ke API terputus
[ingesti] ORD-004 gagal (percobaan 2): koneksi ke API terputus
[ingesti] ORD-004 gagal (percobaan 3): koneksi ke API terputus
[ingesti] berhenti: koneksi ke API terputus pada ORD-004
[ingesti] checkpoint tetap di baris 3; kegagalan dicatat ke catat-galat.log
exit code: 1

Tahap kedua: koneksi diperbaiki (berkas penanda dihapus), lalu skrip yang sama dijalankan ulang tanpa argumen khusus.

rm keadaan_toko/gangguan-aktif
python3 ingestion.py
echo "exit code: $?"

Output yang Dihasilkan:

[ingesti] mulai dari baris ke-4 (lanjut dari checkpoint)
[ingesti] ORD-004 tersimpan
[ingesti] ORD-005 tersimpan
[ingesti] ORD-006 tersimpan
[ingesti] selesai: 3 baris baru, total 6 baris di tabel pesanan
exit code: 0

Terakhir, kita periksa jejak kegagalan dari tahap pertama.

cat keadaan_toko/catat-galat.log

Output yang Dihasilkan:

[gagal] koneksi ke API terputus pada ORD-004; checkpoint tetap di baris 3

Mari kita telaah bagian-bagian kuncinya:

1. Baris 26-33 (ambil_pesanan): kegagalan transien dicoba ulang sampai tiga kali dengan jeda 0,2 detik; setelah semua percobaan habis, fungsi memunculkan ConnectionError agar pemanggil yang memutuskan langkah berikutnya.
2. Baris 64 (for indeks in range(mulai, ...)): perulangan dimulai dari posisi checkpoint, bukan dari nol — inilah resume: tiga baris pertama tidak diproses ulang pada tahap kedua.
3. Baris 66-70 (urutan commit): baris ditulis dengan INSERT OR IGNORE lalu langsung db.commit(); penulisan bersifat idempotent sehingga pemrosesan ulang tidak menghasilkan duplikat.
4. Baris 71 (simpan_checkpoint(indeks + 1)): checkpoint diperbarui hanya setelah commit berhasil — persis aturan urutan yang kita bahas di atas.
5. Baris 74-80 (penangan kegagalan): ConnectionError ditangkap, kegagalan dicatat ke log lengkap dengan posisi checkpoint, dan return 1 memberi exit code non-zero agar scheduler (misalnya cron atau Airflow) tahu eksekusi ini gagal.

Perhatikan perilaku kedua tahap tersebut: tahap pertama berhenti tanpa kehilangan satu baris pun, dan tahap kedua menyelesaikan sisanya hingga total 6 baris — hasil akhir identik dengan pipeline yang selesai tanpa gangguan. Kombinasi klasifikasi, retry aman, checkpoint, dan log kegagalan inilah standar minimum pipeline yang layak berjalan tanpa pengawasan.

Dengan ini perjalanan mengumpulkan data tuntas: file, database, API, webhook, message queue, hingga CDC — lengkap dengan pagination, rate limit, idempotency, validasi, dan penanganan kegagalan. Data yang sudah berada di staging masih mentah dan perlu dibersihkan, dibentuk, lalu digabungkan menjadi model analitik — pekerjaan yang menanti pada Section Data Transformation: ETL dan ELT.