# 機能設計書 27-ステート管理

## 概要

本ドキュメントは、Apache Flinkのステート管理機能（flink-runtime モジュール内StateBackend）の設計仕様を記述する。ステート管理機能は、キー付きステートとオペレーターステートの管理を担当する。

### 本機能の処理概要

StateBackendは、ストリーミングアプリケーションの状態（State）を管理するための抽象レイヤーを提供する。異なる実装により、メモリ内保存やRocksDBなどの外部ストレージへの保存をサポート。

**業務上の目的・背景**：ステートフルなストリーム処理において、ウィンドウ集計、結合、パターン検出などの処理に必要な状態を効率的に管理する。障害復旧時に状態を復元し、Exactly-once処理を保証する。

**機能の利用シーン**：
- キー付きステートの読み書き（ValueState、ListState、MapState等）
- オペレーターステートの管理
- チェックポイント時の状態スナップショット
- 状態の復元とリカバリ

**主要な処理内容**：
1. キー付きステートバックエンドの作成
2. オペレーターステートバックエンドの作成
3. 状態のシリアライゼーション/デシリアライゼーション
4. チェックポイントスナップショットの作成
5. 状態の復元

**関連システム・外部連携**：
- CheckpointCoordinator（チェックポイント）
- CheckpointStorage（状態保存先）
- TaskExecutor（ローカル状態管理）

**権限による制御**：TaskExecutor内でのみアクセス可能。

## 関連画面

本機能は画面機能マッピングには直接含まれていない。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 内部処理として使用 |

## 機能種別

状態管理 / データ保存

## 入力仕様

### 入力パラメータ

#### キー付きステートバックエンド作成

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| parameters | KeyedStateBackendParameters | Yes | 作成パラメータ | null不可 |

#### オペレーターステートバックエンド作成

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| parameters | OperatorStateBackendParameters | Yes | 作成パラメータ | null不可 |

### 入力データソース

- ユーザーコードからの状態アクセス
- チェックポイント復元データ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| CheckpointableKeyedStateBackend | CheckpointableKeyedStateBackend | キー付きステートバックエンド |
| OperatorStateBackend | OperatorStateBackend | オペレーターステートバックエンド |
| KeyedStateHandle | KeyedStateHandle | キー付きステートハンドル |
| OperatorStateHandle | OperatorStateHandle | オペレーターステートハンドル |

### 出力先

- メモリ（HashMapStateBackend）
- RocksDB（RocksDBStateBackend）
- チェックポイントストレージ

## 処理フロー

### 処理シーケンス

```
1. StateBackend初期化
   └─ ジョブ起動時にStateBackend選択

2. キー付きステートバックエンド作成
   └─ createKeyedStateBackend() で作成
   └─ 状態復元（ある場合）

3. 状態アクセス
   └─ getState() でステート取得
   └─ value()/update() で読み書き

4. チェックポイント
   └─ snapshot() で状態スナップショット
   └─ 非同期でストレージに書き込み

5. 状態復元
   └─ チェックポイントから状態読み込み
   └─ ステートバックエンドに復元
```

### フローチャート

```mermaid
flowchart TD
    A[ジョブ起動] --> B[StateBackend選択]
    B --> C{バックエンド種別}
    C -->|HashMap| D[HashMapStateBackend]
    C -->|RocksDB| E[RocksDBStateBackend]
    C -->|ForSt| F[ForStStateBackend]

    D --> G[KeyedStateBackend作成]
    E --> G
    F --> G

    G --> H{復元データあり?}
    H -->|Yes| I[状態復元]
    H -->|No| J[新規作成]

    I --> K[状態アクセス準備完了]
    J --> K

    K --> L[ユーザーコードから状態アクセス]
    L --> M{チェックポイント?}
    M -->|Yes| N[状態スナップショット]
    M -->|No| L
    N --> L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | キー分離 | キー付きステートはキーごとに分離 | キー付きステートアクセス時 |
| BR-002 | 並列度変更 | 状態はキーグループ単位で再配分 | リスケール時 |
| BR-003 | TTL | 設定された時間経過後に状態をクリア | StateTTL有効時 |
| BR-004 | シリアライゼーション | TypeSerializerで状態をシリアライズ | 状態保存時 |

### 計算ロジック

- キーグループの計算（KeyGroupRangeAssignment）
- 状態サイズの推定

## データベース操作仕様

### 操作別データベース影響一覧

本機能は直接データベースを操作しない（RocksDBは組み込みDB）。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | システムエラー | 状態読み書き失敗 | ストレージ確認 |
| StateMigrationException | 互換性エラー | シリアライザ不一致 | スキーマ互換性確認 |
| BackendBuildingException | 設定エラー | バックエンド作成失敗 | 設定確認 |

### リトライ仕様

- 状態読み込み失敗: タスク再起動
- チェックポイント失敗: 再スナップショット

## トランザクション仕様

状態更新はタスク単位でアトミック。チェックポイント完了時に確定。

## パフォーマンス要件

- 状態アクセス: マイクロ秒〜ミリ秒オーダー（バックエンドによる）
- チェックポイント: インクリメンタルCPで効率化

## セキュリティ考慮事項

- 状態データの暗号化（オプション）
- RocksDBローカルファイルのアクセス権限

## 備考

- HashMapStateBackend: 高速だがヒープサイズに制限
- RocksDBStateBackend: 大規模状態に対応
- ForStStateBackend: 次世代バックエンド

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: インターフェースを理解する

StateBackendのインターフェースを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java` | メインインターフェース |
| 1-2 | CheckpointableKeyedStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointableKeyedStateBackend.java` | キー付きステートバックエンド |
| 1-3 | OperatorStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java` | オペレーターステートバックエンド |

**読解のコツ**:
- StateBackendはファクトリとして機能
- **81-91行目**: StateBackendの基本説明（ヒープ/RocksDB）
- **104-105行目**: createKeyedStateBackend()メソッド

#### Step 2: 実装を理解する

各StateBackend実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HashMapStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java` | ヒープ実装 |
| 2-2 | StateBackendLoader.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java` | バックエンド読み込み |

### プログラム呼び出し階層図

```
StateBackend (interface)
    │
    ├─ createKeyedStateBackend(parameters)
    │      └─ CheckpointableKeyedStateBackend
    │             ├─ getState() -> ValueState
    │             ├─ getListState() -> ListState
    │             ├─ getMapState() -> MapState
    │             └─ snapshot() -> KeyedStateHandle
    │
    └─ createOperatorStateBackend(parameters)
           └─ OperatorStateBackend
                  ├─ getListState() -> ListState
                  ├─ getUnionListState() -> ListState
                  └─ snapshot() -> OperatorStateHandle
```

### データフロー図

```
[入力]                    [処理]                           [出力]

ユーザーコード ────────▶ StateBackend ──────────────▶ State値
(状態アクセス)                │
                             ▼
                    KeyedStateBackend
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
     ValueState         ListState          MapState
           │                 │                 │
           └────────────────▼──────────────────┘
                             │
                    チェックポイント時
                             │
                             ▼
                    CheckpointStorage
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java` | ソース | メインインターフェース |
| HashMapStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/hashmap/HashMapStateBackend.java` | ソース | ヒープ実装 |
| StateBackendLoader.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java` | ソース | バックエンド読み込み |
| CheckpointableKeyedStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointableKeyedStateBackend.java` | ソース | キー付きバックエンド |
| OperatorStateBackend.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java` | ソース | オペレーターバックエンド |
| KeyedStateHandle.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateHandle.java` | ソース | キー付きステートハンドル |
