Kafka Streams の State Store 実戦解説: RocksDB と再起動時の復旧

Kafka Streams の状態管理は、集計・結合・ウィンドウなどの有状態処理の肝です。デフォルトで RocksDB をローカルに埋め込み、障害時は changelog トピックから復旧します。

本稿では、RocksDB の基本、再起動時の復旧フロー、スタンバイレプリカ、運用チューニング、試験（CCDAK）で問われやすいポイントを、実務に即して整理します。

State Store の基本と RocksDB の位置づけ

Kafka Streams は、演算の中間結果を State Store に保存します。マテリアライズされた State Store は、同名の changelog トピックで冗長化され、障害時にリプレイして復旧できます。

デフォルトの永続ストアは RocksDB（プロセス内に組み込み）です。オンヒープのメモリストアも選べますが、プロセス終了で内容が消えるため、再起動後は changelog の全リプレイが必要になります。RocksDB はローカルディスクに永続化されるため、チェックポイントが一致していれば増分のみを適用して素早く復旧できます。

• すべての State Store は changelog で冗長化（通常はログ圧縮が有効）
• デフォルトは RocksDB。メモリストアは高速だが再起動コストが高い
• Exactly-once v2 を有効にすると、ストア更新・changelog 書き込み・出力の整合性が担保される

再起動時の復旧フロー（RocksDB）

プロセスが再起動すると、各タスクはローカル state.dir 配下の RocksDB とチェックポイントファイル（changelog の適用済み位置）を確認します。ローカル状態が存在し、チェックポイントとタスク割当が一致すれば、changelog の差分だけを読み込んで整合性を回復し、処理を再開します。

ローカル状態が失われた場合やチェックポイントが不整合な場合は、該当ストアの changelog を先頭からリプレイして完全復旧します。Exactly-once v2 を有効にしていると、トランザクション境界で復旧が打ち切られ、ストアと出力トピックの一貫性が保たれます。協調的リバランス（ステートフルタスクのスティッキー配置）が有効なクラスターでは、同一ホストにタスクを残しやすいためローカル状態の再利用率が上がります。

• 増分復旧の条件: ローカル RocksDB が存在、チェックポイントとタスク割当が一致、破損なし
• 完全復旧のトリガ: ローカル状態の欠落/破損、タスク移動でローカル再利用不可、チェックポイント不整合
• 復旧ソースは常に changelog。入力トピックのオフセットは復旧には使わない

RocksDB ストアの復旧パス

Changelog トピック、ログ圧縮、トンブストーン

各 State Store は key-value の更新を changelog トピックに反映します。changelog はログ圧縮により、最新のキー更新が保持され、古いバージョンは削除されます。削除操作はトンブストーン（キーのみ、値は null）として書き込まれ、復旧時にも削除が正しく反映されます。

正確な復旧のためには、ストアと changelog で同一のシリアライザ/デシリアライザ（Serde）を使うこと、内部トピックのレプリケーション係数を十分に確保することが重要です。RocksDB 側のフラッシュタイミングやキャッシュは復旧の論理には影響しません。復旧は常に changelog の順序に従って適用されます。

• トンブストーンは復旧時も処理され、キーの削除が再現される
• 内部トピックのレプリケーション係数は本番で 3 を推奨（可用性）
• Serde を変更すると復旧に失敗する可能性があるため慎重に移行する

スタンバイレプリカとリバランスで復旧時間を短縮

num.standby.replicas を設定すると、各パーティションの State Store を別ノードでバックグラウンドに保温できます。フェイルオーバ時はスタンバイが Active に昇格し、差分のみの適用で復旧が完了します。代償としてネットワークとストレージのコストがかかります。

協調的リバランスではステートフルタスクのスティッキー配置により、スケールイン/アウト時も同一ホストにタスクが留まりやすく、ローカル state.dir の再利用によって復旧コストが抑えられます。コンテナ運用では state.dir を永続ボリュームに載せることで同様の効果が得られます。

• SLO が厳しい場合は num.standby.replicas=1 以上を検討
• state.dir は高速かつ永続のストレージに配置（例: NVMe + 永続ボリューム）
• スケール操作は協調的リバランスを利用し、タスクのスティッキー性を維持

運用チューニング: state.dir と RocksDB 設定

state.dir は十分な I/O と容量を持つローカル永続ディスクに置き、コンテナならホストボリュームをマウントします。ファイルディスクリプタ上限や I/O スケジューラ設定も重要です。

RocksDB は RocksDBConfigSetter でライトバッファやブロックキャッシュを調整できます。キャッシュ（cache.max.bytes.buffering）とコミット間隔（commit.interval.ms）は、レイテンシと changelog 書き込み量のトレードオフです。復旧スループットは changelog 消費のネットワークとディスク I/Oが支配するため、max.partition.fetch.bytes などのコンシューマ設定も見直します。

Exactly-once v2（processing.guarantee=exactly_once_v2）を推奨。内部トピック（changelog/repartition）のレプリケーション係数を 3 に揃え、障害時のデータ損失を防ぎます。

• state.dir を明示設定し、ノード再配置でもボリュームを引き継ぐ
• 内部トピックの replication.factor と min.insync.replicas を本番水準に
• RocksDB のブロックキャッシュはワーキングセットに見合う容量を割り当て
• commit.interval.ms はビジネス要件に応じてミリ秒〜数秒を検討

設定例（Java Properties + Streams DSL）

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "payments-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
// 永続ディスク配下に配置
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/kafka-streams/state");
// 復旧短縮（コストとトレードオフ）
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
// Exactly-once v2
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// キャッシュとコミット間隔
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 256 * 1024 * 1024L); // 256MB
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 500);
// 内部トピックのレプリケーション（本番は 3 推奨）
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
// 復旧スループットに影響
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 4 * 1024 * 1024); // 4MB

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Payment> payments = builder.stream("payments");
KTable<String, Long> counts = payments
    .groupByKey()
    .count(Materialized.as("payments-counts")); // デフォルトで RocksDB + changelog

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

トラブルシューティングと試験対策ポイント

ローカル state.dir を誤って消すと、完全復旧（changelog 全リプレイ）になり時間を要します。時間短縮にはスタンバイやボリューム永続化、協調的リバランスを組み合わせます。

チェックポイントが壊れている場合は、ログに復旧フォールバックが記録され、先頭からのリプレイに切り替わります。整合性が取れないと判断したら、該当タスクのローカルディレクトリを削除して再生成させるのが確実です。

トポロジ変更でストア名や Serde を変えると、既存の changelog との互換性が崩れ復旧に失敗します。安全に初期化する必要があれば、アプリケーションリセットツールで入力オフセットと内部トピックをクリーンにしてから再デプロイします。

• 復旧は changelog が唯一の真実。入力トピックのオフセットではない
• ローカル状態の再利用は条件付き（チェックポイント一致、破損なし、同一ホスト）
• Serde 互換性と内部トピックの信頼性設定は必須の試験ポイント

問題で確認

よくある質問