Retries dan Timeouts

Jadwal yang benar hanya memastikan workflow dipicu pada waktu yang tepat — tidak menjamin eksekusinya sukses. API sumber data bisa mengalami gangguan 30 detik, koneksi database bisa terputus sesaat, dan sebuah query bisa berjalan jauh lebih lambat dari biasanya. Data engineer membagi kegagalan menjadi dua kategori yang menuntut strategi berlawanan: kegagalan transien yang layak dicoba ulang, dan kegagalan permanen yang justru harus berhenti secepat mungkin. Chapter ini membahas kebijakan retry, backoff, jitter, dan timeout yang membedakan keduanya, lengkap dengan implementasi Python murni.

Kegagalan Transien vs Permanen #

Menghabiskan lima percobaan retry untuk sebuah ValueError akibat schema berubah hanya menunda kegagalan yang pasti terjadi — waktu yang seharusnya dipakai untuk memperbaiki kode. Karena itu retry policy harus dibatasi pada daftar galat transien yang eksplisit, bukan menangkap semua exception sekaligus.

Kebijakan Retry: Backoff dan Jitter #

Tiga angka menentukan kebijakan retry: jumlah percobaan maksimum, jeda awal, dan faktor pengganda. Exponential backoff menggandakan jeda pada setiap kegagalan — misalnya 0,2 detik lalu 0,4 detik lalu 0,8 detik — memberi waktu pemulihan yang makin lama tanpa membebani server sejak awal. Jitter menambahkan komponen acak pada jeda tersebut. Tanpa jitter, puluhan task yang gagal bersamaan akan mencoba ulang pada detik yang persis sama dan membebani server yang sedang pulih; jitter menyebarkan momen percobaan ulang tersebut. Rumusnya ringkas: jeda_n = jeda_awal x faktor^(n-1) + jitter.

Timeout: Batas Durasi Task dan Workflow #

Timeout per task membatasi durasi satu task; timeout per workflow membatasi durasi total sebuah run. Tanpa timeout, task yang macet — misalnya menunggu respons yang tidak pernah datang — bisa menahan slot executor dan menunda seluruh downstream tanpa batas waktu. Praktik umumnya: timeout per task diatur lebih kecil daripada batas workflow secara keseluruhan, sehingga kegagalan terdeteksi dan notifikasi terkirim selagi masih ada waktu bereaksi sebelum SLA terlampaui. Angka yang tepat bergantung pada profil task — batas satu jam untuk query yang biasanya selesai dalam dua menit hanya menunda deteksi kegagalan.

Implementasi Retry Decorator #

Kita implementasikan kebijakan tersebut sebagai decorator yang bisa ditempelkan pada fungsi task apa pun. Fungsi ambil_data_api disimulasikan gagal dua kali karena ConnectionError sebelum sukses; fungsi baca_konfigurasi selalu mengangkat ValueError sebagai contoh kegagalan permanen.

"""Retry decorator: exponential backoff + jitter, hanya untuk galat transien."""
import functools
import random
import time


def retry(maks_percobaan=3, jeda_awal=0.2, faktor=2.0, galat_transien=(ConnectionError,)):
    def dekorator(fn):
        @functools.wraps(fn)
        def pembungkus(*args, **kwargs):
            jeda = jeda_awal
            for percobaan in range(1, maks_percobaan + 1):
                try:
                    return fn(*args, **kwargs)
                except galat_transien:
                    if percobaan == maks_percobaan:
                        raise
                    print(f"  [retry] {fn.__name__} percobaan {percobaan} gagal, tunggu {jeda:.1f} detik")
                    time.sleep(jeda + random.uniform(0, jeda * 0.25))  # jitter
                    jeda *= faktor
        return pembungkus
    return dekorator


random.seed(42)
panggilan = 0


@retry()
def ambil_data_api(sumber):
    global panggilan
    panggilan += 1
    if panggilan <= 2:
        raise ConnectionError(f"koneksi ke {sumber} terputus")
    return f"data dari {sumber}"


@retry()
def baca_konfigurasi(berkas):
    raise ValueError(f"kolom wajib hilang di {berkas}")


print("hasil:", ambil_data_api("api.example.com"))

try:
    baca_konfigurasi("config.yaml")
except ValueError as galat:
    print("gagal permanen (tanpa retry):", galat)

Output yang Dihasilkan:

  [retry] ambil_data_api percobaan 1 gagal, tunggu 0.2 detik
  [retry] ambil_data_api percobaan 2 gagal, tunggu 0.4 detik
hasil: data dari api.example.com
gagal permanen (tanpa retry): kolom wajib hilang di config.yaml

Mari kita telaah baris-baris kode di atas:

1. Baris 7 (parameter kebijakan): seluruh kebijakan hidup di satu tempat — jumlah percobaan, jeda awal, faktor pengganda, dan daftar tipe galat transien yang boleh dicoba ulang.
2. Baris 14: hasil sukses langsung dikembalikan tanpa jeda; percobaan berikutnya tidak diperlukan.
3. Baris 15-17: hanya galat pada galat_transien yang ditangani; percobaan terakhir meneruskan galat ke pemanggil sehingga kegagalan tetap terlihat.
4. Baris 18-20 (backoff + jitter): jeda dicetak lalu dijalankan bersama komponen acak hingga 25% dari jeda; setelah itu jeda digandakan — terlihat pada output: 0,2 detik lalu 0,4 detik.
5. Baris 38-40 dan 45-48: ValueError tidak masuk daftar transien sehingga lolos begitu saja — buktinya tidak ada baris [retry] untuk baca_konfigurasi pada output.

Implementasi Timeout dengan future.result #

Untuk timeout, Python stdlib menyediakan pola bersih lewat concurrent.futures: task dikirim ke thread pool, lalu hasilnya diambil dengan batas waktu eksplisit. Task muat_tabel disimulasikan memakan 1,0 detik sementara batasnya hanya 0,3 detik.

"""Timeout task: membatasi durasi satu task dengan future.result(timeout=...)."""
import concurrent.futures
import time


def muat_tabel(nama_tabel, durasi):
    time.sleep(durasi)  # simulasi query berat
    return f"{nama_tabel}: selesai dimuat"


with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
    tugas = pool.submit(muat_tabel, "pesanan", 1.0)
    try:
        print(tugas.result(timeout=0.3))
    except concurrent.futures.TimeoutError:
        print("TimeoutError: muat_tabel tidak selesai dalam 0.3 detik")

Output yang Dihasilkan:

TimeoutError: muat_tabel tidak selesai dalam 0.3 detik

Mari kita telaah baris-baris kode di atas:

1. Baris 6-8: task simulasi — tidur selama durasi lalu mengembalikan pesan; dalam pipeline nyata, bagian ini adalah query atau pemanggilan API.
2. Baris 12 (pool.submit): task dikirim ke thread pool dan langsung mendapat objek Future; pemanggil tidak menunggu di baris ini.
3. Baris 14 (result(timeout=0.3)): pemanggil menunggu hasil maksimal 0,3 detik — batas waktu per task dalam wujud paling murni.
4. Baris 15-16: batas terlampaui menghasilkan TimeoutError yang bisa ditangkap; workflow kemudian bebas memilih kebijakan lanjutan: retry, notifikasi, atau melewati task.

Retry dan timeout mengurus kegagalan hari ini. Ketika bug telah diperbaiki atau data terlambat akhirnya tiba, periode lampau perlu dijalankan ulang secara menyeluruh — pokok bahasan Chapter Backfill, dan kebijakan lengkap Airflow untuk keduanya terdokumentasi di dokumentasi Apache Airflow .