KAL
Menu

State store

State store lokal dengan RocksDB, changelog topic, dan pemulihan state saat aplikasi restart.

Daftar isi

Pemrosesan stateful menyimpan keadaan antar record sehingga agregasi dapat dibangun secara bertahap, sedangkan broker Kafka hanyalah log yang tidak menyediakan tempat penyimpanan keadaan. Topik ini membahas state store lokal yang menopang operasi stateful pada Kafka Streams, RocksDB sebagai penyimpanan bawaannya, changelog topic yang mencatat setiap perubahan, serta pemulihan keadaan ketika aplikasi restart. Satu simulasi Python memutar state store dan changelog-nya pada aliran kejadian tetap, sehingga keadaan akhir beserta proses pemulihannya dapat diperiksa langsung. Setelah membacanya, penamaan store dan keputusan mengaktifkan atau menonaktifkan pencatatan changelog dapat ditetapkan dengan sadar.

Mengapa pemrosesan stateful membutuhkan state lokal#

Operator stateful, misalnya count setelah groupBy yang dibahas pada topik Primitif stream, memerlukan keadaan yang bertahan antar record. Penghitung per key harus dapat dibaca kembali saat record berikutnya tiba, dan hasil agregasi hanya benar bila penghitung itu tidak hilang di tengah proses. Nilai keadaan tersebut tidak bisa diletakkan pada broker karena broker hanya menyediakan log, bukan pembacaan acak per key. Menyimpannya pada basis data eksternal juga menambah satu lompatan jaringan untuk setiap record yang diproses.

Tempat keadaan itu disimpan disebut state store, yaitu penyimpanan key-value yang menyatu dengan satu task pada instance aplikasi. Setiap task mengurus satu partisi topic input, dan seluruh record ber-key sama berada pada partisi yang sama, sehingga keadaan setiap key cukup dipelihara oleh satu task. Pembacaan dan penulisan ke store berlangsung di memori atau diska mesin aplikasi tanpa menyentuh jaringan, sehingga agregasi berjalan record demi record dengan latensi serendah membaca variabel. Karena store adalah milik task dan bukan milik aplikasi secara keseluruhan, dua instance tidak saling membaca store masing-masing, dan sifat inilah yang menentukan cara keadaannya dicadangkan.

RocksDB sebagai penyimpanan bawaan#

State store bawaan tersedia dalam dua bentuk, yaitu store di dalam memori dan store persisten pada diska. Store persisten menggunakan RocksDB, yaitu mesin penyimpanan key-value yang menulis datanya ke diska lokal, dan bentuk inilah yang direkomendasikan untuk kebanyakan kasus karena kapasitasnya mengikuti diska dan tidak membebani heap aplikasi. Data RocksDB berada di bawah direktori yang ditetapkan lewat konfigurasi state.dir, dengan subfolder per application.id. Store di dalam memori lebih cepat dan tidak menyentuh diska, tetapi ukurannya dibatasi kapasitas heap sehingga hanya cocok untuk keadaan yang kecil.

Bentuk store mengikuti bentuk datanya. Tersedia store key-value untuk agregasi biasa, store berwindow untuk agregasi windowing, store session untuk pengelompokan sesi, serta varian timestamped dan versioned yang menyimpan riwayat waktu per key. Pada DSL, bentuk dan nama store dinyatakan lewat Materialized, dan agregasi berwindow otomatis menggunakan store berwindow tanpa deklarasi tambahan. Potongan berikut menamai store hasil count dengan nama eksplisit.

java Contoh: menamai state store hasil agregasi dengan Materialized 
// tidak-dijalankan
KTable<String, Long> jumlahKata = kata.groupBy(
        (kunci, teks) -> teks)
    .count(Materialized.as("jumlah-kata"));

Tanpa Materialized, Kafka Streams tetap membuat store internal dengan nama yang dihasilkan otomatis, sehingga memberi nama secara eksplisit memudahkan pencarian topic cadangannya serta pengukuran ukurannya. Serdes key dan value pada store mengikuti konfigurasi aplikasi atau disebutkan eksplisit bila tipe datanya berbeda. Store yang dibuat lewat DSL berperilaku sama dengan store yang dibangun langsung lewat Processor API, karena keduanya akhirnya berupa implementasi StateStore yang sama.

Changelog topic mencatat setiap perubahan#

Store yang fault-tolerant didukung oleh satu topic changelog, yaitu topic yang mencatat setiap perubahan keadaan dengan key yang sama seperti store-nya. Setiap kali task melakukan put ke store, perubahan itu juga ditulis ke changelog, sehingga topic tersebut memuat riwayat lengkap keadaan store. Pencatatan ini diaktifkan secara bawaan pada store yang dibuat lewat DSL, dan nama topicnya mengikuti pola <application.id>-<nama-store>-changelog.

Changelog dikompaksi sebagaimana pola changelog yang dijelaskan pada modul Desain topic, yaitu broker menyimpan nilai terkini per key dan menghilangkan nilai yang sudah digantikan. Kompaksi menjaga topic agar tidak tumbuh tanpa batas, mengurangi storage pada cluster, dan mempercepat pemulihan karena record yang perlu diputar ulang menjadi lebih sedikit. Changelog untuk store berwindow menggunakan kombinasi delete,compact karena key-nya menyertakan batas window, sehingga window yang sudah kedaluwarsa ikut terhapus.

Pencatatan changelog diatur lewat StoreBuilder pada Processor API. Metode withLoggingEnabled menerima konfigurasi tambahan untuk topic-nya, misalnya min.insync.replicas, sedangkan withLoggingDisabled mematikan pencadangan sepenuhnya.

java Contoh: StoreBuilder dengan changelog dan konfigurasinya 
// tidak-dijalankan
StoreBuilder<KeyValueStore<String, Long>> pembuatStore =
    Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("jumlah-kata"),
            Serdes.String(), Serdes.Long())
        .withLoggingEnabled(Map.of("min.insync.replicas", "1"));

Pemulihan state dan standby replica#

Ketika aplikasi restart atau task berpindah ke instance lain, RocksDB di instance baru masih kosong. Sebelum pemrosesan dilanjutkan, Kafka Streams memutar ulang changelog dari awal dan menerapkan setiap record ke store, sehingga isi store kembali persis seperti sebelum kegagalan. Kompaksi menyisakan satu nilai terkini per key, sehingga durasi pemulihan terutama ditentukan banyaknya key dan bukan panjang riwayat perubahannya. Proses pemulihan (restore) berjalan otomatis dan tidak mengubah logika pemrosesan sedikit pun.

Diagram berikut merangkum kedua arah tersebut, yaitu setiap put yang ditulis ke changelog dan pemutaran ulang changelog saat pemulihan berlangsung.

record topic inputtask Streamsstate store lokalRocksDBbroker Kafkatopic changelogterkompaksiput() menulisrestore: changelog diputar ulang
State store lokal didukung changelog topic pada broker

Pemulihan dari nol memakan waktu, dan Kafka Streams menyediakan jalan pintas berupa standby replica, yaitu salinan state store yang dipelihara di instance lain dan terus diperbarui dari changelog. Jumlah salinannya diatur lewat konfigurasi num.standby.replicas. Apabila task harus berpindah, penjadwalan memilih instance yang sudah memegang salinan mutakhir, sehingga pemulihan hanya mengejar selisih kecil alih-alih memutar seluruh changelog dari awal. Standby menambah konsumsi diska dan jaringan, tetapi memperpendek jeda pemrosesan saat kegagalan terjadi.

Simulasi state store lokal dan changelog-nya#

Simulasi berikut memutar penghitung per key pada enam kejadian tetap menggunakan Python murni, tanpa broker dan tanpa RocksDB. Sebuah dictionary Python berperan sebagai state store yang menyimpan keadaan terkini per key, sedangkan daftar berperan sebagai changelog yang mencatat setiap perubahan. Aliran kejadiannya sengaja dipilih sama dengan contoh groupBy dan count pada topik Primitif stream agar hubungan keduanya terlihat. Skrip juga menyimulasikan compaction pada changelog beserta pemulihan state dari hasil compaction-nya.

python simulasi-state-store.py 
# simulasi state store lokal dengan changelog, tanpa broker dan tanpa RocksDB

# kejadian tetap dari satu partisi topic, meniru hasil groupBy + count
# pada topik Primitif stream: setiap kejadian menaikkan penghitung per key
kejadian = ["log", "kejadian", "log", "dibaca", "log", "kejadian"]

state = {}      # state store lokal: penghitung terkini per key
changelog = []  # changelog topic: setiap perubahan dicatat berurutan


print("proses record satu per satu:")
for offset, key in enumerate(kejadian):
    lama = state.get(key, 0)
    baru = lama + 1
    state[key] = baru                       # put ke state store
    changelog.append((offset, key, baru))   # tulis ke changelog topic
    print(f"  offset={offset} key={key}: count {lama} -> {baru}"
          " (put state, tulis changelog)")

print()
print("state store lokal (keadaan terkini per key):")
for key in sorted(state):
    print(f"  {key} -> {state[key]}")

print()
print(f"changelog ({len(changelog)} record sebelum compaction):")
for offset, key, nilai in changelog:
    print(f"  offset={offset} key={key} count={nilai}")

# compaction menyimpan satu nilai terkini per key agar restore cepat
kompaksi = {}
for offset, key, nilai in changelog:
    kompaksi[key] = (offset, nilai)
print()
print("changelog setelah compaction (satu record terkini per key):")
urutan_kompaksi = sorted(kompaksi.items(), key=lambda pasangan: pasangan[1][0])
for key, (offset, nilai) in urutan_kompaksi:
    print(f"  offset={offset} key={key} count={nilai}")

# task restart: RocksDB kosong, state dibangun ulang dari changelog
print()
print("restore: RocksDB kosong, putar changelog dari awal")
state_pulih = {}
urutan_restore = sorted(kompaksi.items(), key=lambda pasangan: pasangan[1][0])
for key, (offset, nilai) in urutan_restore:
    state_pulih[key] = nilai
    print(f"  restore key={key} count={nilai} (dari offset={offset})")
print("state pulih sama dengan state akhir:", state_pulih == state)
text Keluaran 
proses record satu per satu:
  offset=0 key=log: count 0 -> 1 (put state, tulis changelog)
  offset=1 key=kejadian: count 0 -> 1 (put state, tulis changelog)
  offset=2 key=log: count 1 -> 2 (put state, tulis changelog)
  offset=3 key=dibaca: count 0 -> 1 (put state, tulis changelog)
  offset=4 key=log: count 2 -> 3 (put state, tulis changelog)
  offset=5 key=kejadian: count 1 -> 2 (put state, tulis changelog)

state store lokal (keadaan terkini per key):
  dibaca -> 1
  kejadian -> 2
  log -> 3

changelog (6 record sebelum compaction):
  offset=0 key=log count=1
  offset=1 key=kejadian count=1
  offset=2 key=log count=2
  offset=3 key=dibaca count=1
  offset=4 key=log count=3
  offset=5 key=kejadian count=2

changelog setelah compaction (satu record terkini per key):
  offset=3 key=dibaca count=1
  offset=4 key=log count=3
  offset=5 key=kejadian count=2

restore: RocksDB kosong, putar changelog dari awal
  restore key=dibaca count=1 (dari offset=3)
  restore key=log count=3 (dari offset=4)
  restore key=kejadian count=2 (dari offset=5)
state pulih sama dengan state akhir: True

Enam kejadian menghasilkan tiga kelompok pada state store, yaitu dibaca bernilai satu, kejadian bernilai dua, dan log bernilai tiga. Changelog memuat enam record karena setiap perubahan tercatat, sedangkan hasil compaction hanya menyisakan tiga record terkini per key. Bagian pemulihan menunjukkan bahwa state store kosong yang dibangun ulang dari changelog hasil compaction berakhir dengan keadaan yang sama persis dengan keadaan akhir. Pada topologi nyata, daftar itu adalah topic terkompaksi bernama <application.id>-<nama-store>-changelog, dan pemutaran ulangnya dilakukan oleh task sebelum record pertama diproses. Nama store yang eksplisit memudahkan pencarian topic tersebut di antara topic internal aplikasi yang jumlahnya ikut bertambah.

Kesalahan umum#

  1. Menganggap state store dapat dibaca lintas instance. Store lokal hanya menyimpan keadaan partisi yang dipegang task-nya, sehingga instance lain tidak melihat isi store tersebut. Akses keadaan dari luar dilakukan lewat fitur interactive queries atau dengan membaca topic hasil pemrosesannya, bukan dengan berbagi folder state.
  2. Menonaktifkan changelog demi menghemat topic. Store tanpa changelog tidak dapat dipulihkan dan tidak dapat memiliki standby replica, sehingga satu kali restart membuat seluruh keadaan agregasi mulai dari nol. Nonaktifkan pencatatan hanya untuk keadaan yang memang dapat dibangun ulang dengan cepat dari topic input-nya.
  3. Mengabaikan kapasitas diska pada state.dir. RocksDB menyimpan keadaan di diska lokal, dan store berwindow menambah ukurannya karena setiap window memegang agregatnya sendiri. Ukur laju pertumbuhan keadaan terhadap diska yang tersedia sebelum menaikkan ukuran window atau jumlah partisi.
  4. Menyimpan keadaan pada koleksi aplikasi, misalnya dictionary statis di luar topologi. Keadaan semacam itu tidak ditulis ke changelog sehingga tidak pulih saat task berpindah, dan nilainya tidak dapat dipantau dari luar. Seluruh keadaan yang dibutuhkan agregasi dideklarasikan melalui state store.

Ringkasan#

  • Pemrosesan stateful menyimpan keadaan pada state store lokal per task, sehingga agregasi per key berjalan tanpa lompatan jaringan ke broker atau basis data eksternal.
  • Store persisten bawaan menggunakan RocksDB di bawah state.dir, dengan varian key-value, berwindow, session, timestamped, dan versioned.
  • Store yang fault-tolerant ditulis ke changelog topic terkompaksi bernama <application.id>-<nama-store>-changelog, dan topic itulah yang diputar ulang untuk mengembalikan keadaan saat aplikasi restart.
  • Konfigurasi num.standby.replicas memelihara salinan store di instance lain untuk memperpendek pemulihan saat task berpindah.

Referensi#