Changelog Topic #

Dalam sistem terdistribusi stateful, tantangan terbesar kita bukanlah bagaimana memproses data saat kondisi normal, melainkan bagaimana memulihkan keadaan (state) aplikasi secara instan ketika terjadi bencana (disaster recovery). Jika sebuah wadah mikroservis yang menyimpan data stateful di disk lokalnya (RocksDB) tiba-tiba hancur akibat kegagalan server fisik, bagaimana kita bisa menjamin bahwa data saldo berjalan atau preferensi profil tidak lenyap? Di Apache Kafka, jaminan ketahanan keadaan ini dikelola secara otomatis melalui komponen internal bernama Changelog Topic (Topik Catatan Perubahan). Melalui artikel ini, kita akan membedah secara mendalam peran penting Changelog Topic, hubungan sinkronisasi 1:1 antara mesin database RocksDB lokal dengan broker Kafka, alur pemulihan bencana (Disaster Recovery Replay Path), perbandingan dengan topik biasa, mitigasi skenario kegagalan pemulihan, serta optimasi pemulihan menggunakan konfigurasi Standby Replicas dan parameter retensi.

Ketahanan State Store melalui Changelog Topic #

Di dalam arsitektur Kafka Streams, local state store (seperti RocksDB) dirancang untuk berkinerja tinggi. Namun, RocksDB lokal ini bersifat mudah menguap (volatile) dalam skala kluster: ia terikat langsung pada sistem penyimpanan disk lokal VM server tempat aplikasi kita dijalankan.

• Bahaya Ephemeral Storage: Jika kita mendeploy aplikasi kita di Kubernetes menggunakan Cloud Provider (AWS EKS, GCP GKE) dengan volume penyimpanan non-persisten (ephemeral storage), setiap kali pod aplikasi berpindah node, seluruh data direktori RocksDB lokal di /tmp/kafka-streams akan dihapus bersih.
• Solusi Changelog: Untuk mencegah malapetaka kehilangan data ini, Kafka Streams secara otomatis membuat topik Kafka internal khusus untuk setiap state store yang kita deklarasikan. Setiap kali aplikasi mengubah data di RocksDB lokal (baik melalui operasi put, delete, maupun update), aplikasi juga akan memancarkan record perubahan tersebut secara asinkron ke topik changelog yang bersangkutan di broker Kafka.

Karakteristik Topik Changelog #

Topik changelog dibuat secara otomatis oleh Kafka Streams dengan beberapa penyetelan default yang sangat ketat untuk menjamin keamanan data:

• Log Compaction: Topik changelog dikonfigurasi dengan kebijakan pembersihan cleanup.policy=compact. Ini memastikan broker Kafka hanya menyimpan status record terbaru untuk setiap kunci, menghemat ruang penyimpanan di broker sambil mempertahankan riwayat state yang utuh.
• Replication Factor Tinggi: Secara default, jika tidak diatur lain, topik changelog akan dibuat dengan Replication Factor bernilai 3 (di lingkungan produksi) untuk memastikan state store kita aman dari crash broker tunggal.
• Min In-Sync Replicas: Properti seperti min.insync.replicas=2 memastikan penulisan event changelog dianggap sukses hanya jika telah terduplikasi ke minimal dua broker.

Perbandingan Changelog Topic vs Topik Kafka Biasa #

Sangat penting bagi kita untuk memahami perbedaan mendasar antara topik changelog internal dengan topik Kafka biasa yang kita buat secara manual:

Menulis secara manual ke dalam topik changelog menggunakan Kafka Producer eksternal adalah anti-pattern yang sangat fatal, karena akan merusak state RocksDB lokal kita secara diam-diam dan memicu inkonsistensi saat proses inisialisasi ulang berjalan.

Alur Pemulihan Bencana (Disaster Recovery Replay Path) #

Ketika sebuah instance aplikasi kita mengalami crash permanen dan Kubernetes menjadwalkan ulang pod tersebut di server baru, database RocksDB lokal akan kosong melompong. Pada saat inilah Disaster Recovery aktif untuk membangun kembali RocksDB dari nol.

Langkah-langkah Reconstruct State #

Proses pemulihan ini berjalan dalam urutan logis berikut:

flowchart TD
    Step1["1. Pod Baru Aktif<br>Mendeteksi Folder RocksDB Lokal Kosong"] --> Step2["2. Task Berstatus RECOVERING<br>Memulai Konsumsi dari Changelog Topic"]
    Step2 --> Step3["3. Replay Event<br/>Membaca Event dari Offset 0 hingga LSO (Latest Offset)"]
    Step3 --> Step4["4. Rekonstruksi<br>Menulis Ulang Setiap Pasangan Key-Value ke RocksDB Lokal"]
    Step4 --> Step5["5. Status RUNNING<br>Transisi Siap Memproses Data Aliran Masukan Utama"]

Selama fase pemulihan ini (status RECOVERING), StreamTask yang bersangkutan tidak memproses data baru dari topik input utama. Hal ini dilakukan untuk mencegah ketidaksesuaian data (data corruption). Konsumen internal menangguhkan pembacaan input, fokus melakukan replay data dari topik changelog ke RocksDB lokal secepat mungkin. Setelah posisi offset konsumen changelog mencapai offset terujung (Latest Stable Offset / LSO) di broker, barulah task tersebut beralih ke status RUNNING dan mulai memproses input stream kembali.

sequenceDiagram
    autonumber
    participant App as Pod KStreams Baru
    participant Broker as Broker Kafka (Changelog)
    participant Disk as RocksDB Lokal (Disk)

    Note over App, Disk: "Fase 1: Deteksi Cold Start"
    App->>Disk: Cek folder state lokal
    Disk-->>App: Folder kosong / Korup!
    
    Note over App, Broker: "Fase 2: Proses Replay (State Restoration)"
    App->>Broker: Subscribe ke Topic changelog-store
    App->>Broker: Tarik data dari Offset 0 (awal)
    
    loop Replay Log
        Broker-->>App: Kirim Batch Event (Key-Value)
        App->>Disk: Tulis cepat ke RocksDB lokal
    end

    Note over App, Disk: "Fase 3: Transisi Status Run"
    Broker-->>App: Mencapai Offset Terujung (LSO)
    App->>App: Ubah status Task dari RECOVERING ke RUNNING
    App->>Disk: Siap melayani Query REST & Pemrosesan Input

Menghindari Jeda Booting dengan Standby Replicas #

Meskipun alur pemulihan di atas menjamin tidak adanya kehilangan data (zero data loss), ia memiliki satu kekurangan fatal: Cold Start Latency.

• Jika ukuran database RocksDB kita berukuran puluhan gigabyte (misal 50GB), proses mengunduh dan melakukan replay 50GB data dari broker Kafka ke server lokal melalui jaringan dapat memakan waktu puluhan menit hingga berjam-jam. Selama waktu tersebut, task akan membeku (lagging) dan throughput bisnis terganggu.

Solusi Terbaik: Standby Replicas #

Untuk mengatasi masalah dinginnya waktu start ini, Kafka Streams menyediakan fitur Standby Replicas (num.standby.replicas).

• Cara Kerja: Jika kita menyetel num.standby.replicas=1, Kafka Streams akan membuat replika cadangan pasif (standby task) di instance aplikasi kedua untuk setiap task aktif yang berjalan di instance aplikasi pertama.
• Sinkronisasi Real-Time: Standby task ini tidak memproses data input utama, namun ia terus-menerus membaca topik changelog secara real-time dan menerapkannya ke database RocksDB lokal cadangannya sendiri.
• Failover Instan (Hot Standby): Jika instance pertama yang memegang task aktif tiba-tiba mati, instance kedua dapat langsung mempromosikan standby task miliknya menjadi active task dalam hitungan milidetik karena data RocksDB lokalnya sudah 99.9% sinkron dengan active task yang mati. Tidak ada jeda download jaringan yang lama.

Kode Penerapan: Anti-Pattern vs Solusi Rebuilding Aman #

Mari kita pelajari contoh bagaimana mengonfigurasi opsi logging changelog, standby replicas, dan integrasi dengan probe Kubernetes menggunakan StateListener secara tepat di Java SDK.

Anti-Pattern: Mematikan Changelog demi Latensi Semu #

Mencoba mematikan fitur changelog pada State Store persisten RocksDB hanya untuk meningkatkan throughput penulisan sedikit adalah kesalahan fatal di lingkungan produksi.

// ANTI-PATTERN: Menurunkan performa ketahanan dengan menonaktifkan changelog secara permanen.
// ✗ Jika server pod crash, seluruh data state store akan musnah selamanya tanpa bisa dipulihkan.
public class VulnerableTopologyBuilder {
    public static void build(StreamsBuilder builder) {
        // Mendefinisikan persistent store tanpa logging
        StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("my-store"),
            Serdes.String(),
            Serdes.String()
        )
        // ✗ SANGAT BERBAHAYA: Mematikan Changelog! Data tidak akan di-backup ke Broker Kafka.
        .withLoggingDisabled(); 

        builder.addStateStore(storeBuilder);
    }
}

Solusi Praktis: Mengaktifkan Changelog dan Mengonfigurasi Standby Replicas #

Berikut adalah konfigurasi yang direkomendasikan untuk produksi. Kita secara eksplisit mengaktifkan pencatatan changelog, mengonfigurasi parameter topik internal secara kustom, dan mengaktifkan Standby Replicas.

// BENAR: Mengonfigurasi State Store dengan logging aktif, custom changelog parameters, dan Standby Replicas.
// ✓ Menjamin durabilitas maksimum dan waktu failover yang instan di Kubernetes.
public class HighAvailabilityTopology {
    
    public static Properties createStreamsConfiguration() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "finance-ledger-service");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        // ✓ MENGAKTIFKAN STANDBY REPLICAS
        // Menugaskan 1 task replika cadangan di host lain untuk failover milidetik
        props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
        
        // Optimasi consumer internal untuk membaca changelog lebih agresif
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), 1000);
        
        return props;
    }

    public static void buildSecureTopology(StreamsBuilder builder) {
        // Mengonfigurasi parameter kustom untuk topik changelog internal di broker
        Map<String, String> changelogConfig = new HashMap<>();
        // ✓ Memaksa minimal ISR = 2 untuk keamanan replikasi di broker
        changelogConfig.put("min.insync.replicas", "2");
        // Mengurangi batas ukuran segmen agar kompresi log compaction berjalan lebih sering
        changelogConfig.put("segment.bytes", "67108864"); // 64 MB

        StoreBuilder<KeyValueStore<String, String>> secureStoreBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("secure-balance-store"),
            Serdes.String(),
            Serdes.String()
        )
        // ✓ MENGAKTIFKAN LOGGING CHANGELOG (Default aktif, namun kita kustomisasi parameternya)
        .withLoggingEnabled(changelogConfig);

        builder.addStateStore(secureStoreBuilder);
    }
}

Integrasi Liveness/Readiness Probe Kubernetes dengan StateListener #

Agar load balancer Kubernetes tidak mengirimkan trafik API (melalui Interactive Queries) ke pod kita saat pod tersebut sedang dalam status memulihkan changelog, kita harus mendaftarkan listener untuk memantau status aplikasi:

// BENAR: Menggunakan StateListener untuk mengelola status kesiapan aplikasi (Readiness Probe).
public class KStreamsReadinessService {
    private volatile boolean isReady = false;

    public void registerListener(KafkaStreams streams) {
        streams.setStateListener((newState, oldState) -> {
            log.info("Transisi status Kafka Streams dari {} ke {}", oldState, newState);
            
            // ✓ Pod dianggap siap hanya jika berada dalam status RUNNING
            // Jika statusnya REBALANCE atau RECOVERING, isReady diubah menjadi false
            if (newState == KafkaStreams.State.RUNNING) {
                isReady = true;
            } else {
                isReady = false;
            }
        });
    }

    public boolean isAppReady() {
        return isReady;
    }
}

Skenario Kegagalan Khusus: Mengapa Pemulihan Changelog Bisa Gagal? #

Di lingkungan produksi yang kompleks, ada beberapa skenario di mana alur pemulihan changelog kita bisa mengalami kegagalan fatal:

1. Data Key Null pada Topik Changelog #

• Masalah: RocksDB menyimpan data dalam format key-value. Jika ada record masukan di topik changelog yang memiliki kunci null, RocksDB akan melemparkan eksepsi kegagalan penulisan saat proses pemulihan berjalan, menyebabkan task lumpuh selamanya (hanging task).
• Mitigasi: Pastikan semua upstream producer menggunakan partisi dan kunci yang valid. Hindari pengiriman event tanpa key ke topik yang dikonfigurasi dengan KTable/Changelog.

2. Inkonsistensi Skema (Schema Mismatch) #

• Masalah: Jika skema data (misalnya objek Avro atau Protobuf) dirubah di upstream tanpa mengikuti aturan kompatibilitas mundur (backward compatibility), proses deserialisasi event changelog lama dari offset 0 saat pemulihan akan melemparkan SerializationException.
• Mitigasi: Selalu gunakan Confluent Schema Registry dengan aturan kompatibilitas penuh (Full Compatibility) dan lakukan pengujian schema migration di staging sebelum deploy ke produksi.

Mengelola Ukuran dan Kebijakan Pembersihan Changelog #

Di lingkungan produksi berskala besar, jika kita tidak memantau topik changelog, kita dapat mengalami penumpukan penyimpanan disk di broker. Berikut adalah beberapa taktik operasi untuk mengelolanya:

1. Mempercepat Siklus Log Compaction #

Secara default, log cleaner thread di broker Kafka bertugas memangkas record lama yang kuncinya terduplikasi. Kita dapat mempercepat siklus pembersihan ini dengan menyetel parameter broker berikut:

• log.cleaner.min.cleanable.ratio (Default 0.5): Turunkan menjadi 0.2 agar broker melakukan pembersihan segment log saat 20% data di dalamnya sudah berstatus kotor (dirty records).
• log.cleaner.delete.retention.ms (Default 86400000 ms / 24 jam): Jika kita mengirimkan record null (tombstone), kurangi waktu penahanan penanda hapus ini agar disk cepat dibersihkan.

2. Monitoring Status Restorasi State #

Aplikasi kita dapat memantau status pemulihan task dengan mendaftarkan kelas StateRestoreListener:

kafkaStreams.setGlobalStateRestoreListener(new StateRestoreListener() {
    @Override
    public void onRestoreStart(TopicPartition topicPartition, String storeName, long startOffset, long endOffset) {
        log.info("Memulai restorasi store [{}] dari partisi {}. Target offset: {}", 
            storeName, topicPartition.partition(), endOffset);
    }

    @Override
    public void onBatchRestored(TopicPartition topicPartition, String storeName, long batchEndOffset, long numRestored) {
        // Logging kemajuan restorasi berkala
    }

    @Override
    public void onRestoreEnd(TopicPartition topicPartition, String storeName, long totalRestored) {
        log.info("✓ Restorasi selesai untuk store [{}]. Total data dipulihkan: {}", storeName, totalRestored);
    }
});

Ringkasan #

• Changelog Topic — Topik internal di broker Kafka yang didesain khusus untuk merekam setiap pembaruan data state store secara asinkron untuk disaster recovery.
• 1:1 Mapping — Setiap instance local state store (RocksDB) terikat langsung dengan satu partisi dari topik changelog internal yang bersertifikasi log-compaction.
• Replay Path — Alur pemulihan keadaan di mana pod baru mengunduh dan menulis ulang key-value record dari topik changelog Kafka ke RocksDB kosong dari offset awal hingga LSO.
• Standby Replicas — Instansi cadangan pasif yang menduplikasi state store aktif secara real-time via changelog untuk menjamin failover secepat kilat tanpa jeda cold-start.
• StateListener — Pustaka Java SDK untuk mendeteksi perubahan status internal KStreams (RUNNING, REBALANCE, RECOVERING) guna diintegrasikan dengan liveness/readiness probe Kubernetes.
• Log Compaction — Kebijakan pembersihan segment topik di broker yang memangkas entri lama dengan kunci terduplikasi, hanya menyisakan status data keadaan terbaru.

← Sebelumnya: State Store   Berikutnya: Fault Tolerance →