State Store #

Dalam merancang arsitektur mikroservis pemrosesan data real-time, salah satu masalah paling rumit yang sering kita hadapi adalah bagaimana mengelola penyimpanan keadaan (state) saat melakukan komputasi stateful. Operasi seperti melakukan agregasi pendapatan per jam, menghitung frekuensi klik pengguna, atau men-join-kan dua aliran event sangat bergantung pada kemampuan sistem untuk menyimpan dan membaca keadaan secara cepat. Apache Kafka menyediakan abstraksi penyimpanan bernama State Store untuk memecahkan masalah ini. State Store di Kafka Streams menggabungkan kecepatan akses memori lokal melalui engine RocksDB dengan ketahanan data terdistribusi. Melalui artikel ini, kita akan membedah secara mendalam arsitektur internal State Store, alur tulis-baca data (write-read paths) yang melibatkan JVM Heap Cache, RocksDB dan Changelog Topic, teknik tuning performa RocksDB secara manual, serta bagaimana memanfaatkan Interactive Queries (IQ) untuk mengekspos data keadaan lokal ke dunia luar secara aman tanpa memicu penurunan latensi.

Peran Penting Local State Store dalam Pemrosesan Stateful #

Ketika kita melakukan pemrosesan data aliran tradisional, database eksternal terpusat (seperti Redis, PostgreSQL, atau Cassandra) sering kali digunakan sebagai wadah penyimpanan keadaan. Namun, pendekatan ini memiliki keterbatasan skalabilitas yang sangat besar pada throughput tinggi:

  • Network Latency Overhead: Setiap record data yang diproses harus melakukan panggilan jaringan (RPC) bolak-balik ke database eksternal. Jika kita memproses 50.000 record per detik, beban jaringan ini akan memicu bottleneck parah.
  • Tantangan ACID: Mengelola konsistensi data transaksional terdistribusi lintas sistem (Kafka dan database eksternal) sangatlah sulit, terutama jika terjadi kegagalan jaringan di tengah pemrosesan.

Solusi Kafka: Local State Store #

Untuk menyingkirkan hambatan latensi jaringan ini, Kafka Streams menerapkan prinsip menyatukan komputasi dengan data. Setiap instance aplikasi kita menyimpan potongan data keadaannya secara lokal di dalam mesin penyimpanan tertanam (Local State Store).

  • Akses Zero-Network: Saat StreamTask melakukan filter atau update saldo akun berjalan, ia menulis saldo baru tersebut ke disk lokal atau memori RAM dari mesin server tempat aplikasi itu sendiri berjalan. Latensinya adalah mikrodetik, bukan milidetik.
  • Isolasi Task: Setiap State Store secara ketat hanya memuat data untuk partisi topik input yang ditugaskan ke StreamTask tersebut. Tidak ada perebutan kunci (locking contention) antar thread pemroses yang berbeda.

Mengapa RocksDB? Pemrosesan Stateful di Luar Heap JVM #

Secara default, Kafka Streams menggunakan RocksDB sebagai mesin penyimpanan persisten lokal untuk State Store. RocksDB adalah database key-value bertanam (embedded key-value store) buatan Meta (Facebook) yang ditulis dalam bahasa C++ dan sangat dioptimalkan untuk media penyimpanan SSD cepat.

Pencegahan GC Pause Panjang (Off-Heap Memory) #

Salah satu alasan utama mengapa RocksDB dipilih, bukan HashMap Java biasa di dalam memori JVM, adalah untuk menghindari dampak Garbage Collection (GC) Pause yang parah.

  • Masalah JVM Heap: Jika kita menyimpan jutaan data profil pengguna di dalam memori Heap JVM menggunakan objek Java biasa, Garbage Collector JVM harus memindai jutaan objek tersebut setiap kali melakukan siklus pembersihan. Ini akan menghentikan seluruh utas eksekusi aplikasi kita (Stop-the-World GC Pause) selama beberapa detik hingga menit, yang merusak jaminan latensi real-time.
  • Solusi Off-Heap RocksDB: RocksDB menyimpan datanya di luar memori Heap JVM (Off-Heap Memory). RocksDB beroperasi langsung di ruang memori kernel sistem operasi dan menggunakan teknik LSM-Tree (Log-Structured Merge-tree) untuk menulis data ke disk secara berurutan. Data hanya dikonversi dari biner ke objek Java sesaat ketika dibutuhkan oleh logika aplikasi, menjaga Heap JVM tetap bersih dan ringan.

Sinergi Penulisan Data: Alur Tulis dan Baca State Store #

Untuk menyeimbangkan antara kecepatan baca-tulis memori dengan ketahanan data fisik dari kegagalan listrik, Kafka Streams menerapkan alur pemrosesan tiga lapis yang saling terintegrasi:

  1. JVM Heap Cache (Lapis Kecepatan) Setiap State Store memiliki cache memori internal di dalam JVM Heap (dapat dikonfigurasi melalui cache.max.bytes.buffering). Saat record baru masuk, data ditulis ke dalam cache ini terlebih dahulu untuk deduplikasi cepat di memori.
  2. RocksDB Storage Engine (Lapis Persisten Lokal) Secara berkala (atau saat cache memori JVM penuh), data akan diturunkan (flushed) ke instance database RocksDB lokal yang tersimpan di SSD server.
  3. Changelog Topic (Lapis Durabilitas Terdistribusi) Pada saat yang sama ketika data ditulis ke RocksDB, perubahan tersebut dikirimkan sebagai event changelog ke topik internal di broker Kafka (application-id-store-name-changelog). Topik ini bertindak sebagai asuransi jiwa bagi data keadaan lokal kita.
flowchart TD
    subgraph APP ["Aplikasi Java KStreams (StreamTask)"]
        direction TB
        Logika["Logika Bisnis (e.g. aggregate)"]
        JVMCache["JVM Heap Cache (Memory Buffering)"]
    end

    subgraph LOCAL ["Penyimpanan Fisik Lokal (Host SSD)"]
        RocksDB[("RocksDB Store (Off-Heap C++)")]
    end

    subgraph BROKER ["Infrastruktur Broker Kafka"]
        Changelog["Topic: application-id-user-store-changelog"]
    end

    Logika -->|"1. Tulis cepat"| JVMCache
    JVMCache -->|"2. Flush berkala (Off-Heap)"| RocksDB
    JVMCache -.->|"3. Replikasi Asinkron (Record Update)"| Changelog

    style Logika stroke:#0288d1,stroke-width:2px
    style JVMCache stroke:#388e3c,stroke-width:2px
    style RocksDB stroke:#f57c00,stroke-width:2px
    style Changelog stroke:#d32f2f,stroke-width:2px

Kustomisasi dan Tuning Performa RocksDB #

Meskipun RocksDB sangat tangguh, konfigurasi default dari Kafka Streams terkadang tidak optimal untuk semua jenis beban kerja komputasi. Untuk skenario dengan volume penulisan super tinggi, RocksDB bawaan dapat mengalami perlambatan akibat pembagian blok memori (compaction style). Kita dapat menyesuaikan RocksDB menggunakan implementasi RocksDBConfigSetter.

Berikut adalah contoh kustomisasi parameter memori RocksDB untuk meminimalkan jeda waktu flush disk dan memperbesar kapasitas block cache:

// BENAR: Mengimplementasikan RocksDBConfigSetter untuk mengoptimalkan alokasi memori off-heap RocksDB.
// ✓ Mengontrol block cache, write buffer size, dan menghindari kemacetan penulisan ke disk (write stalls).
public class CustomRocksDbConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        // 1. Buat filter block policy untuk mempercepat pencarian data dengan Bloom Filter
        org.rocksdb.BlockBasedTableConfig tableConfig = new org.rocksdb.BlockBasedTableConfig();
        
        // Atur ukuran block cache (misal: 64MB)
        tableConfig.setBlockCache(new org.rocksdb.LRUCache(64 * 1024 * 1024L));
        // Atur ukuran blok data (default 4KB)
        tableConfig.setBlockSize(4 * 1024L);
        // Tambahkan bloom filter dengan 10 bit per key (mengurangi disk read palsu)
        tableConfig.setFilterPolicy(new org.rocksdb.BloomFilter(10, false));
        
        options.setTableFormatConfig(tableConfig);

        // 2. Optimalkan Write Buffer (MemTable) size
        // writeBufferSize menentukan batas memori per partisi sebelum di-flush ke disk (misal: 16MB)
        options.setWriteBufferSize(16 * 1024 * 1024L);
        // Jumlah maksimum write buffer yang menumpuk di memori sebelum diblokir
        options.setMaxWriteBufferNumber(3);

        // 3. Konfigurasi kompresi data untuk menghemat ruang penyimpanan SSD lokal
        options.setCompressionType(CompressionType.LZ4_COMPRESSION);
    }

    @Override
    public void close(String storeName, Options options) {
        // Bersihkan resource jika ada
    }
}

Untuk mendaftarkan konfigurasi kustom di atas ke dalam aplikasi Kafka Streams kita, kita cukup menambahkannya ke properti konfigurasi:

Properties props = new Properties();
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDbConfig.class.getName());

Kode Penerapan: Anti-Pattern vs Solusi dengan State Store Resmi #

Mari kita bandingkan implementasi berbahaya yang sering ditemui dengan cara yang benar menggunakan State Store terkelola di Kafka Streams.

Skenario Kasus Penggunaan #

Kita ingin melacak jumlah login gagal beruntun dari pengguna e-commerce (failed-login-events). Jika pengguna gagal login lebih dari 3 kali berturut-turut, kita ingin menandai pengguna tersebut untuk verifikasi tambahan.

Anti-Pattern: Menggunakan HashMap InMemory Sederhana Tanpa State Store Resmi # 

// ANTI-PATTERN: Menggunakan Map memori internal Java di dalam pemrosesan aliran.
// ✗ Rentan kehilangan seluruh data saat crash, tidak didukung backup Changelog, dan memicu OutOfMemory.
public class VolatileMemoryProcessor {
    // ✗ BAHAYA: Map ini akan hilang saat restart dan tidak terdistribusi lintas task
    private static final Map<String, Integer> failedLoginCounts = new ConcurrentHashMap<>();

    public static void processStream(KStream<String, String> stream) {
        stream.foreach((userId, eventJson) -> {
            int currentCount = failedLoginCounts.getOrDefault(userId, 0) + 1;
            failedLoginCounts.put(userId, currentCount);

            if (currentCount >= 3) {
                log.warn("User {} has exceeded 3 failed login attempts!", userId);
            }
        });
    }
}

Solusi Praktis: Mendefinisikan Persistent State Store Terkelola #

Di bawah ini adalah implementasi yang benar dengan mendaftarkan State Store persisten berbasis RocksDB yang aman dari crash.

// BENAR: Menggunakan KStream.process() untuk mengakses State Store persisten RocksDB.
// ✓ Aman dari crash, terhubung ke Changelog Topic untuk replikasi status otomatis.
public class SecureStateStoreProcessor implements Processor<String, String, String, String> {
    private KeyValueStore<String, Integer> failedLoginStore;
    private ProcessorContext<String, String> context;

    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context;
        // ✓ Mengambil referensi ke RocksDB store terdaftar
        this.failedLoginStore = context.getStateStore("failed-login-store");
    }

    @Override
    public void process(Record<String, String> record) {
        String userId = record.key();
        
        // Membaca count login gagal saat ini dari RocksDB lokal secara aman
        Integer currentCount = failedLoginStore.get(userId);
        if (currentCount == null) {
            currentCount = 0;
        }

        int newCount = currentCount + 1;
        
        // Simpan nilai baru ke RocksDB lokal
        failedLoginStore.put(userId, newCount);

        if (newCount >= 3) {
            // Loloskan alert transaksi mencurigakan ke downstream
            context.forward(new Record<>(userId, "BLOCKED", record.timestamp()));
        }
    }

    @Override
    public void close() { }
}

Interactive Queries (IQ): Mengakses State Secara Langsung #

Salah satu fitur paling revolusioner di Kafka Streams adalah Interactive Queries (IQ). Biasanya, saat aplikasi menulis data hasil olahan ke database, aplikasi lain harus melakukan query ke database eksternal tersebut. Dengan IQ, State Store di dalam aplikasi Kafka Streams dapat ditanyakan secara langsung oleh layanan eksternal melalui REST API tanpa perlu menulis data ke database eksternal terlebih dahulu.

Namun, karena data KTable kita terbagi secara lokal di beberapa partisi dan instansi, kita harus melakukan pencarian dua tahap:

  1. Menentukan instansi mana yang memegang partisi untuk kunci (key) yang sedang dicari.
  2. Melakukan panggilan HTTP REST ke instansi tersebut jika kunci tidak berada di memori lokal instansi saat ini.

Kode Penerapan: Mengkueri State Store via Interactive Queries #

Berikut adalah contoh kode Java untuk menemukan lokasi instansi metadata dan mengambil data dari State Store lokal.

// BENAR: Mengimplementasikan Interactive Queries untuk mengekspos data State Store lokal.
// ✓ Menggunakan metadata untuk menentukan lokasi partisi kunci sebelum query dibaca.
public class InteractiveQueryService {
    private final KafkaStreams streams;

    public InteractiveQueryService(KafkaStreams streams) {
        this.streams = streams;
    }

    public Integer getFailedLoginCount(String userId) {
        // Hubungkan pengecekan liveness streams terlebih dahulu
        if (streams.state() != KafkaStreams.State.RUNNING) {
            throw new IllegalStateException("Kluster Kafka Streams sedang dalam status rebalance atau idle.");
        }

        // Langkah 1: Cari tahu broker metadata mana yang memegang partisi key tersebut
        KeyQueryMetadata metadata = streams.queryMetadataForKey(
            "failed-login-store",
            userId,
            Serdes.String().serializer()
        );

        if (metadata == null) {
            throw new RuntimeException("Metadata tidak ditemukan untuk key: " + userId);
        }

        // Langkah 2: Evaluasi apakah partisi key tersebut ada di instansi aktif saat ini
        if (metadata.activeHost().port() == getCurrentInstancePort()) {
            // ✓ Key berada di memori lokal, baca langsung dari RocksDB lokal tanpa jaringan
            ReadOnlyKeyValueStore<String, Integer> store = streams.store(
                StoreQueryParameters.fromNameAndType(
                    "failed-login-store",
                    QueryableStoreTypes.readOnlyKeyValueStore()
                )
            );
            return store.get(userId);
        } else {
            // ✓ Key berada di instansi lain, arahkan HTTP Request ke host tujuan
            String targetUrl = String.format("http://%s:%d/users/%s/failed-logins",
                metadata.activeHost().host(),
                metadata.activeHost().port(),
                userId
            );
            return fetchFromRemoteInstance(targetUrl);
        }
    }

    private int getCurrentInstancePort() { return 8080; }
    private Integer fetchFromRemoteInstance(String url) { return 0; /* Simulasi REST call */ }
}

Ringkasan #

  • Local State Store — Antarmuka penyimpanan data keadaan lokal terintegrasi yang menyatukan data dengan komputasi untuk menghindari latensi panggilan jaringan eksternal.
  • RocksDB Engine — Database key-value bertanam berbasis LSM-Tree yang beroperasi di luar memori heap JVM (off-heap) untuk menghindari GC Pause Stop-the-world.
  • Write-Read Path — Perubahan state ditulis pertama kali ke JVM cache, disusul flush ke RocksDB lokal dan backup asinkron ke Changelog Topic di broker Kafka.
  • Changelog Topic — Topik internal berreplikasi tinggi yang merekam setiap transisi data state store untuk menjamin pemulihan data lokal saat kegagalan node.
  • RocksDBConfigSetter — Antarmuka khusus di Java untuk mengustomisasi detail parameter memori RocksDB off-heap seperti block cache size dan compression type.
  • Interactive Queries (IQ) — Mekanisme query langsung ke State Store lokal aplikasi dari REST API luar tanpa perlu memindahkan data ke database terpisah.
  • num.standby.replicas — Konfigurasi replika state store pasif pada instance lain guna mempercepat pemulihan keadaan tanpa jeda waktu cold start recovery.

← Sebelumnya: Topology vs Processor   Berikutnya: Changelog Topic →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact