Idempotensi pada Bagian 7 menuntut satu hal dari hilir: event_id harus bisa dipercaya. Kepercayaan itu runtuh bila event yang masuk berubah bentuk — field hilang saat tim frontend refactor, nilai total menjadi string, atau producer lain sudah upgrade ke schema v2 dengan field baru. Pada arsitektur event-driven, producer dan consumer dikembangkan tim yang berbeda dan rilis pada waktu berbeda, sehingga kontrak event harus dinyatakan eksplisit dan diperiksa di pintu masuk. Bagian ini memasang gerbang tersebut: schema schema_pesanan_v1.json sebagai kontrak, validator ringan berbasis stdlib sebagai pemeriksa, dan aturan versi yang menolak event dari masa depan tanpa menolak field baru. Prinsip kontrak data dibedah pada chapter Schema Validation dan Data Contracts; di sini kita menerapkannya pada stream.

Kontrak dalam Berkas Schema #

Schema disimpan sebagai berkas JSON di config/schema_pesanan_v1.json — dialek JSON Schema ringan yang mendukung type, required, properties, enum, pattern, minimum, dan format: iso-datetime:

json 
{
  "nama": "event_pesanan",
  "versi": 1,
  "type": "object",
  "required": [
    "event_id",
    "schema_version",
    "event_time",
    "pesanan_id",
    "pelanggan_id",
    "wilayah",
    "total",
    "status"
  ],
  "properties": {
    "event_id": {
      "type": "string",
      "pattern": "^evt-\\d{6}$"
    },
    "schema_version": {
      "type": "integer"
    },
    "event_time": {
      "type": "string",
      "format": "iso-datetime"
    },
    "pesanan_id": {
      "type": "string",
      "pattern": "^ORD-\\d{4,}$"
    },
    "pelanggan_id": {
      "type": "string",
      "pattern": "^PLG-\\d{3,}$"
    },
    "wilayah": {
      "type": "string",
      "enum": ["jakarta", "bandung", "surabaya", "medan", "makassar"]
    },
    "total": {
      "type": "integer",
      "minimum": 0
    },
    "status": {
      "type": "string",
      "enum": ["baru", "dibayar", "dikirim", "selesai", "dibatalkan"]
    }
  }
}

Field versi di puncak berkas menandai versi kontrak; field schema_version pada tiap event harus cocok dengannya. Identitas diikat pola regex — evt-\d{6}, ORD-\d{4,}, PLG-\d{3,} — dan domain nilai diikat enum sehingga typo seperti "dikembalikan" tertangkap sebelum mengotori hilir. Satu aturan lainnya longgar dengan sengaja: field tambahan di luar properties diterima, karena itulah yang membuat schema kompatibel ke depan.

Validator dan Pemeriksaan Versi #

Fungsi validasi_event di src/validator.py memeriksa versi paling awal, baru menjalankan aturan bidang:

python 
def validasi_event(event, schema: dict, path: str = "event") -> list[str]:
    """Validasi satu event; kembalikan daftar pesan galat (kosong = valid).

    Versi diperiksa lebih dahulu: ``schema_version`` event harus cocok
    dengan ``versi`` schema agar event dari producer yang sudah upgrade
    tertolak dengan jelas, bukan divalidasi dengan aturan yang keliru.
    """
    galat: list[str] = []
    versi_schema = schema.get("versi")
    versi_event = event.get("schema_version") if isinstance(event, dict) else None
    if versi_schema is not None and versi_event != versi_schema:
        galat.append(
            f"{path}: schema_version {versi_event!r} tidak cocok dengan "
            f"schema versi {versi_schema}"
        )
    galat += _validasi(event, schema, path)
    return galat

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

  1. Baris 59-60 (versi_schema, versi_event): versi diambil dari dua sumber — schema yang dimuat consumer dan event yang dikirim producer; event non-dictionary ditangani aman dengan None.
  2. Baris 61-65 (pemeriksaan versi): ketidakcocokan versi menjadi galat tersendiri sebelum aturan bidang dijalankan. Tanpa pemeriksaan ini, event v2 akan divalidasi dengan aturan v1 — mungkin lolos, mungkin gagal setengah — dan galatnya menyesatkan.
  3. Baris 66 (_validasi(...)): pemeriksaan rekursif aturan dialek — tipe, required, enum, pattern, minimum, format ISO 8601 — yang setiap galatnya menyertakan lokasi field (misal event.status) agar mudah ditelusuri.

Fungsi validasi_batch memproses daftar sekaligus dan hanya mengembalikan entri bermasalah beserta indeksnya — pola yang sama dengan validator proyek Proyek: Pipeline Ingestion API.

Menjalankan Validasi #

Tiga event bermasalah khas dunia nyata diuji terhadap schema v1 (kutipan demo_validator.py):

python 
schema = muat_schema("config/schema_pesanan_v1.json")
events = buat_pesanan_events(20, seed=42)
print(f"validasi 20 event dari producer: {len(validasi_batch(events, schema))} bermasalah")

# Tiga event bermasalah khas di dunia nyata
rusak1 = dict(events[0]); rusak1.pop("total")          # field wajib hilang
rusak2 = dict(events[1]); rusak2["status"] = "aneh"    # enum tidak dikenal
rusak3 = dict(events[2]); rusak3["schema_version"] = 2 # producer lain upgrade
rusak3["kode_promo"] = "DISKON10"

for nama, event in (("tanpa total", rusak1), ("status aneh", rusak2), ("versi 2", rusak3)):
    print(f"\nevent {nama}: {serialisasi_event(event).decode('utf-8')[:80]}...")
    for galat in validasi_event(event, schema):
        print(f"  - {galat}")

Output yang Dihasilkan:

text 
validasi 20 event dari producer: 0 bermasalah

event tanpa total: {"event_id":"evt-000001","schema_version":1,"event_time":"2024-03-01T09:00:09+00...
  - event: field wajib 'total' tidak ada

event status aneh: {"event_id":"evt-000002","schema_version":1,"event_time":"2024-03-01T09:00:12+00...
  - event.status: nilai 'aneh' tidak ada di enum ['baru', 'dibayar', 'dikirim', 'selesai', 'dibatalkan']

event versi 2: {"event_id":"evt-000003","schema_version":2,"event_time":"2024-03-01T09:00:32+00...
  - event: schema_version 2 tidak cocok dengan schema versi 1

Dua puluh event dari simulator lolos tanpa galat, dan ketiga event bermasalah tertolak dengan pesan yang menyebut field serta alasannya. Di pipeline lengkap, event yang gagal validasi tidak diteruskan ke agregasi — pola pada test_validasi_menyaring_event_rusak_sebelum_hilir memasukkan event bermasalah ke posisi ke-11, memvalidasi, lalu hanya meneruskan 10 event valid ke topic. Pada produksi, event tertolak biasanya dialihkan ke topic karantina (dead letter queue) agar bisa diperiksa dan diproses ulang — bukan dibuang.

Versi dan Evolusi Kontrak #

Aturan versi di validator ini sengaja konservatif: consumer v1 menolak event schema_version: 2 apa pun isinya. Alasannya pragmatic — consumer belum tahu arti field baru, menerima event itu berarti menebak. Evolusi kontrak yang sehat berjalan dua arah: producer menambah field opsional (v1.1) yang consumer lama abaikan dengan aman karena field tambahan memang diterima, lalu perubahan yang menghapus atau mengubah makna field menaikkan versi mayor dan menuntut consumer upgrade. Perjalanan penuh kompatibilitas maju-mundur dibedah pada chapter Schema Evolution dan Schema Registry.

Gerbang validasi menuntaskan sisi keandalan: event valid, tidak dobel, urut per pelanggan. Satu pertanyaan terakhir yang wajib dijawab pipeline produksi: bagaimana kita tahu semuanya masih sehat saat stream berjalan? Bagian penutup membangun pemantauannya.