Fault Tolerance #

Ketika kita mengoperasikan sistem pemrosesan aliran data terdistribusi dalam skala besar di lingkungan produksi, kegagalan bukanlah masalah “apakah” akan terjadi, melainkan “kapan” akan terjadi. Server fisik bisa mati mendadak, jaringan antarsel dapat mengalami pemotongan sementara, container pod Kubernetes bisa terkena evicted akibat kehabisan memori, atau thread pemroses bisa lumpuh akibat eksepsi yang tidak terduga. Untuk menghadapi tantangan kestabilan ini, Apache Kafka Streams menyediakan mekanisme Fault Tolerance (Toleransi Kesalahan) tingkat lanjut. Mekanisme ini mampu memulihkan tugas pemrosesan secara otomatis tanpa intervensi manual dan tanpa mengorbankan konsistensi data. Melalui artikel ini, kita akan membedah arsitektur internal toleransi kesalahan di Kafka Streams, pembagian beban kerja ke dalam StreamTask, siklus hidup task state machine, perbedaan peran active dan standby tasks, cara kerja protokol Cooperative Sticky Rebalance, deteksi kegagalan via Heartbeat Timeout, penanganan kegagalan thread, dan konfigurasi ketahanan produksi.

Pembagian Kerja: Relasi Partisi Terhadap StreamTask #

Untuk memahami bagaimana Kafka Streams menangani kegagalan, kita harus memahami bagaimana unit pemrosesan dibagi secara logis. Kafka Streams menggunakan abstraksi bernama StreamTask sebagai unit paralelisme terkecil.

• Formula Penentuan Jumlah Task: Jumlah StreamTask ditentukan langsung oleh jumlah partisi maksimal dari topik input yang dibaca oleh aplikasi. Jika aplikasi kita membaca dari Topic_A (memiliki 4 partisi) dan Topic_B (memiliki 4 partisi), Kafka Streams akan membuat tepat 4 StreamTask (Task 0, Task 1, Task 2, dan Task 3).
• Isolasi Tugas Stateful: Setiap StreamTask bertindak sebagai mesin pemroses mini yang mandiri. Task memelihara RocksDB local state store sendiri dan memiliki offset konsumen tersendiri. Isolasi ini sangat penting: jika Task 2 mengalami crash, Task 0, 1, dan 3 tetap dapat berjalan memproses data di thread/host masing-masing tanpa terganggu.

Siklus Hidup StreamTask (Task State Machine) #

Setiap StreamTask di dalam Kafka Streams dikelola melalui sebuah mesin status (state machine) internal yang ketat. Transisi status ini sangat penting untuk memastikan tidak ada kebocoran data atau duplikasi saat kegagalan terjadi:

• CREATED: Status awal ketika task baru saja dibentuk di memori. Pada tahap ini, task sedang mendaftarkan partisi input dan menginisialisasi pustaka RocksDB lokal.
• RESTORING: Jika task bersifat stateful dan mendeteksi bahwa data RocksDB lokalnya kosong atau tidak lengkap, ia akan masuk ke status RESTORING. Di sini, task mengaktifkan StateRestoreListener dan melakukan replay data dari topik changelog Kafka.
• RUNNING: Setelah inisialisasi atau restorasi selesai 100%, task bertransisi ke status RUNNING. Ini adalah satu-satunya status di mana task aktif membaca dari topik input utama dan memproses logika bisnis.
• PAUSED: Jika konsumen hilir mengalami overload (backpressure) atau terjadi rebalance tengah jalan, task dapat dijeda sementara (PAUSED) untuk menunda pemrosesan record baru.
• CLOSED: Status akhir ketika aplikasi dimatikan secara bersih (close()) atau ketika task harus dipindahkan ke instance lain akibat rebalance. Saat ditutup, semua offset komit terakhir akan ditulis secara sinkron ke broker, dan database RocksDB lokal akan ditutup dengan aman.

Peran Active Task vs Standby Task #

Ketika kita mendeploy aplikasi Kafka Streams di beberapa instance VM atau kontainer, tugas-tugas (tasks) ini akan disebar ke seluruh instansi yang aktif untuk beban kerja terdistribusi:

1. Active Task (Tugas Aktif) #

Active Task adalah tugas yang secara aktif mengonsumsi data dari topik input Kafka, mengeksekusi logika topologi pemrosesan, memperbarui state store RocksDB lokal, dan menuliskan hasilnya ke topik output.

2. Standby Task (Tugas Siaga) #

Standby Task adalah replika bayangan pasif dari Active Task yang berjalan di instance terpisah.

• Tugas: Standby Task tidak mengonsumsi data dari topik input utama dan tidak memproses logika transformasi. Satu-satunya tugasnya adalah terus-menerus mengonsumsi topik changelog dari broker untuk memperbarui replika database RocksDB lokalnya agar selalu sinkron dengan Active Task.
• Tujuan: Mempercepat pemulihan bencana (Hot Standby Failover). Jika instance pembawa Active Task mati, Standby Task di instance lain dapat langsung dipromosikan menjadi Active Task tanpa perlu melewati fase cold start restoration (mengunduh gigabyte data dari broker via jaringan).

SKENARIO FAILOVER TUGAS:
-------------------------------------------------------------------
Kondisi Awal (Normal):
  Instance A ──> Memegang Active Task 0 (Menulis ke RocksDB_0 & Changelog)
  Instance B ──> Memegang Standby Task 0 (Membaca Changelog ──> Update RocksDB_0_Backup)

Kondisi Gagal (Instance A Crash):
  Instance B ──> Deteksi Kehilangan Instance A via Heartbeat Timeout
  Instance B ──> Mempromosikan Standby Task 0 menjadi Active Task 0 secara instan!
-------------------------------------------------------------------

Protokol Cooperative Sticky Rebalance #

Rebalance adalah proses mendistribusikan ulang alokasi tugas (StreamTasks) ke seluruh instansi aplikasi yang masih hidup. Pada versi-versi lama Kafka, rebalance menggunakan protokol Eager Rebalance (pola stop-the-world) yang sangat lambat karena menghentikan seluruh pemrosesan data di semua instansi selama proses rebalance berlangsung.

Sejak Kafka 2.4, Kafka Streams menggunakan protokol Cooperative Sticky Rebalance sebagai default. Protokol ini membawa peningkatan ketahanan yang luar biasa melalui dua pilar utama:

1. Bersifat Sticky (Lengket) #

Saat terjadi perubahan jumlah instansi (misal, satu pod mati), protokol ini akan mempertahankan alokasi task pada instansi yang masih hidup sebisa mungkin. Task tidak akan dipindahkan dari host-nya saat ini jika tidak benar-benar diperlukan.

2. Bersifat Cooperative (Kooperatif) #

Alih-alih menghentikan seluruh task di semua instansi secara serempak, Cooperative Rebalance membagi proses pemindahan task menjadi beberapa fase bertahap (Incremental Cooperative Rebalancing):

• Fase A: Instansi yang memelihara task yang akan dipindahkan melepaskan kepemilikan task tersebut secara sukarela. Task lain pada instansi lain tetap berjalan memproses data seperti biasa.
• Fase B: Instansi baru mengambil alih task yang dilepaskan tadi dan memulai inisialisasi state store.
• Pendekatan incremental ini memastikan bahwa aplikasi kita tetap dapat melayani trafik komputasi selama proses rebalance berjalan, menyingkirkan fenomena lag parah (rebalance storms).

flowchart TD
    subgraph EAGER ["Eager Rebalance (Stop-the-world)"]
        direction TB
        E1["Crash / Scale-Out Terdeteksi"] --> E2["Hentikan SEMUA pemrosesan di semua instance"]
        E2 --> E3["Hitung ulang alokasi tugas dari nol"]
        E3 --> E4["Mulai ulang pemrosesan di semua instance"]
    end

    subgraph COOPERATIVE ["Cooperative Sticky Rebalance (Incremental)"]
        direction TB
        C1["Crash / Scale-Out Terdeteksi"] --> C2["Tentukan task yang HARUS bermigrasi"]
        C2 --> C3["Lepas HANYA task yang bermigrasi, task lain tetap RUNNING"]
        C3 --> C4["Assign task migrasi ke instance tujuan secara bertahap"]
    end

    style E2 stroke:#d32f2f,stroke-width:2px
    style C3 stroke:#388e3c,stroke-width:2px

Tabel Perbandingan Protokol Rebalance #

Mekanisme Deteksi Kegagalan via Heartbeat Timeout #

Untuk mendeteksi apakah salah satu instansi aplikasi kita mati, Kafka Streams mengandalkan thread internal bernama Heartbeat Thread yang berjalan secara mandiri di setiap konsumen internal task.

• Heartbeat Pengiriman: Thread ini secara konstan mengirimkan sinyal ping (heartbeat) ke broker yang bertindak sebagai Group Coordinator. Kecepatan ping dikonfigurasi melalui properti heartbeat.interval.ms (default 3 detik).
• Session Timeout: Jika broker Group Coordinator tidak menerima sinyal heartbeat dari suatu instansi selama lebih dari durasi session.timeout.ms (default 45 detik), broker akan mengasumsikan bahwa instansi tersebut telah mati secara fisik. Broker kemudian akan menginisialisasi rebalance untuk mendistribusikan ulang task yang hilang ke instansi yang masih hidup.
• Max Poll Interval: Berbeda dengan crash fisik, jika thread pemroses utama terhambat oleh proses GC Pause yang lama atau logika bisnis yang lambat, ia akan gagal memanggil loop poll() internal. Properti max.poll.interval.ms (default 5 menit) membatasi waktu tunggu pemanggilan loop ini. Jika terlampaui, instansi tersebut akan melepaskan diri dari grup konsumen dan memicu rebalance.

Penanganan Kegagalan Thread (Uncaught Exceptions Handling) #

Secara default di Kafka Streams, jika sebuah thread pemroses (StreamThread) melempar eksepsi runtime yang tidak tertangkap (seperti NullPointerException atau SerializationException akibat format data buruk), thread tersebut akan mati.

• Masalah: Jika semua thread pemroses di dalam instansi aplikasi mati satu per satu, aplikasi kita akan menjadi “zombie”: proses Java tetap berjalan di OS (sehingga Kubernetes liveness probe menganggap pod hidup), namun tidak ada data yang diproses sama sekali.

Anti-Pattern: Menggunakan Blok Catch Fragil di Setiap Operator #

Membungkus setiap operator logika DSL dengan blok try-catch manual secara berulang adalah boilerplate code yang buruk dan tidak menangani kegagalan internal JVM.

// ANTI-PATTERN: Membungkus logika pemrosesan dengan try-catch ad-hoc secara manual.
// ✗ Sangat tidak rapi, rawan melewatkan eksepsi tingkat rendah, dan tidak menangani kematian utas eksekusi.
public class FragileErrorHandling {
    public static void build(StreamsBuilder builder) {
        builder.<String, String>stream("input-events")
            .mapValues(value -> {
                try {
                    // Logika bisnis yang rentan error
                    return processJson(value);
                } catch (Exception e) {
                    // ✗ Penanganan ad-hoc lokal yang tidak melaporkan status crash ke orchestrator
                    log.error("Error diproses!", e);
                    return null; 
                }
            });
    }
    private static String processJson(String in) { return ""; }
}

Solusi Praktis: Menggunakan UncaughtExceptionHandler Resmi #

Sejak Apache Kafka 2.8, kita dapat mengonfigurasi penangan kesalahan tingkat utas secara global menggunakan StreamsUncaughtExceptionHandler. Ini memungkinkan aplikasi kita mengambil keputusan strategis saat terjadi kegagalan.

// BENAR: Menggunakan StreamsUncaughtExceptionHandler untuk mengelola kematian thread secara otomatis.
// ✓ Mendukung opsi REPLACE_THREAD untuk mengganti thread mati dengan thread baru secara otomatis,
// atau SHUTDOWN_CLIENT untuk mematikan aplikasi agar dideploy ulang oleh Kubernetes.
public class ResilientStreamsApp {
    
    public static void main(String[] args) {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "resilient-analytics-service");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input-events").to("output-events");

        KafkaStreams streams = new KafkaStreams(builder.build(), config);

        // ✓ MENETAPKAN UNCAUGHT EXCEPTION HANDLER GLOBAL
        streams.setUncaughtExceptionHandler(exception -> {
            log.error("Eksepsi fatal tidak tertangani di utas pemroses!", exception);
            
            // Opsi A: Ganti thread yang mati dengan thread baru tanpa mematikan aplikasi
            // Cocok untuk transient errors seperti database connection drop sementara
            if (exception.getCause() instanceof TransientException) {
                log.info("Mengganti utas pemroses yang mati...");
                return StreamThreadExceptionResponse.REPLACE_THREAD;
            }

            // Opsi B: Matikan klien secara bersih (Shutdown Client)
            // Ini akan mematikan proses JVM, memicu Kubernetes untuk menyalakan pod baru (self-healing)
            log.warn("Mematikan instansi aplikasi secara bersih agar di-restart oleh orchestrator...");
            return StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
        });

        streams.start();
    }

    static class TransientException extends RuntimeException {}
}

Konfigurasi Parameter Ketahanan Produksi #

Untuk memastikan toleransi kesalahan berjalan optimal di lingkungan produksi, berikut adalah beberapa parameter konfigurasi penting yang harus kita setel secara preventif:

1. num.standby.replicas (Default 0) #

• Rekomendasi: Setel menjadi 1 untuk state store berukuran besar (>5GB) guna menjamin failover milidetik tanpa jeda waktu restorasi.

2. acceptable.recovery.lag (Default 10000 records) #

• Rekomendasi: Setel batas lag restorasi maksimum yang diizinkan untuk standby task sebelum ia boleh dideklarasikan siap dipromosikan menjadi active task.

3. task.timeout.ms (Default 300000 ms / 5 menit) #

• Rekomendasi: Menentukan berapa lama task aktif diizinkan membeku atau tidak merespons broker sebelum dianggap mati dan dipindahkan ke instance lain.

Properties props = new Properties();
// Aktifkan Standby Replicas
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
// Toleransi lag komit maksimum untuk hot standby (misal 5000 records)
props.put(StreamsConfig.ACCEPTABLE_RECOVERY_LAG_CONFIG, 5000L);
// Batas waktu timeout task dikurangi menjadi 1 menit untuk deteksi kegagalan lebih cepat
props.put(StreamsConfig.TASK_TIMEOUT_MS_CONFIG, 60000L);

Ringkasan #

• Fault Tolerance — Mekanisme pemulihan mandiri terdistribusi di Kafka Streams untuk mendeteksi, mengisolasi, dan memulihkan kegagalan tanpa kehilangan keadaan (state).
• StreamTask Partitioning — Paralelisme diisolasi di tingkat partisi melalui StreamTask, memastikan kegagalan satu partisi tidak menghentikan partisi lainnya.
• Task State Machine — Pengelolaan siklus hidup task melalui status transisi terstruktur (CREATED, RESTORING, RUNNING, PAUSED, CLOSED) demi keamanan offset.
• Standby Task — Replika bayangan pasif yang terus menyinkronkan state via changelog untuk melakukan Hot Standby Failover milidetik saat active task mengalami crash.
• Cooperative Sticky Rebalance — Protokol redistribusi alokasi task bertahap (incremental) tanpa menghentikan pemrosesan data pada task yang tidak bermigrasi.
• Heartbeat Thread — Thread latar belakang konsumen yang mengirim ping berkala ke Group Coordinator untuk mengonfirmasi status keaktifan node.
• REPLACE_THREAD — Sinyal respons dari global Exception Handler untuk menghidupkan kembali utas Java pemroses yang mati akibat eksepsi runtime secara otomatis.
• SHUTDOWN_CLIENT — Respons penanganan error untuk mematikan proses JVM secara bersih sehingga orkestrator (Kubernetes) dapat mendeteksi kegagalan dan menyalakan pod baru.

← Sebelumnya: Changelog Topic   Berikutnya: Windowing →