# 機能設計書 36-ChangelogStateBackend

## 概要

本ドキュメントは、Apache FlinkのChangelogStateBackend機能について記載する。この機能は、ステート変更をログに記録することで、チェックポイント間隔を短縮し、高速なリカバリを実現するステートバックエンドラッパーである。

### 本機能の処理概要

ChangelogStateBackendは、委譲先のステートバックエンド（HashMapまたはRocksDB）をラップし、すべてのステート変更をChangelogに記録する。これにより、チェックポイント時に完全なスナップショットを取得する必要がなくなり、差分（Changelog）のみを永続化することでチェックポイントを高速化する。

**業務上の目的・背景**：大規模ステートを持つストリーム処理ジョブでは、チェックポイントの完了に時間がかかることが課題である。ChangelogStateBackendは、ステート変更を継続的にログに記録し、定期的なマテリアライゼーションと組み合わせることで、チェックポイント時間を大幅に短縮する。これにより、より頻繁なチェックポイントが可能になり、障害時のデータロスを最小化できる。

**機能の利用シーン**：
- チェックポイント間隔の短縮が必要な場合
- 大規模ステートを持つストリーム処理ジョブ
- 障害時のリカバリ時間を短縮したい場合
- Exactly-once保証が必要な高頻度チェックポイントシナリオ

**主要な処理内容**：
1. 委譲先ステートバックエンドのラップ
2. ステート変更のChangelogへの記録
3. 差分チェックポイントの実行
4. 定期的なマテリアライゼーション（ステートの実体化）
5. Changelog truncation（不要になったログの削除）

**関連システム・外部連携**：
- StateChangelogStorage（DFS DSTL等）
- 委譲先StateBackend（HashMap/RocksDB）
- CheckpointStorage
- PeriodicMaterializationManager

**権限による制御**：特定のロールによる制御なし。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面とは直接関連しない（バックエンド処理） |

## 機能種別

状態管理 / チェックポイント最適化 / 差分永続化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| state.changelog.enabled | boolean | No | Changelog有効化 | true/false |
| state.changelog.storage | String | No | Changelogストレージタイプ | 有効なストレージ名 |
| execution.checkpointing.changelog.periodic-materialize.enabled | boolean | No | 定期マテリアライゼーション有効化 | true/false |
| execution.checkpointing.changelog.periodic-materialize.interval | Duration | No | マテリアライゼーション間隔 | 正の期間 |
| state.changelog.materialization.max-failures-allowed | int | No | マテリアライゼーション最大失敗回数 | 正の整数 |

### 入力データソース

- Flink設定ファイル（flink-conf.yaml）
- ExecutionConfig

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ChangelogStateHandle | Handle | 差分ステートハンドル |
| MaterializedSnapshot | Handle | マテリアライズされたスナップショット |
| ChangelogStateBackendHandle | Handle | Changelog+Materialized統合ハンドル |

### 出力先

- StateChangelogStorage（Changelogデータ）
- CheckpointStorage（マテリアライズデータ）

## 処理フロー

### 処理シーケンス

```
1. ChangelogStateBackend初期化
   └─ 委譲先StateBackendをラップ

2. ChangelogKeyedStateBackend作成
   └─ StateChangelogWriterを取得

3. ステート操作時
   └─ 委譲先に操作を委譲
   └─ StateChangeLoggerで変更をログ記録

4. チェックポイント時
   └─ Changelogの差分を永続化
   └─ マテリアライズ済みスナップショット参照

5. 定期マテリアライゼーション
   └─ 委譲先バックエンドのフルスナップショット
   └─ Changelog truncation

6. リストア時
   └─ マテリアライズスナップショットをリストア
   └─ Changelog変更を再適用
```

### フローチャート

```mermaid
flowchart TD
    A[ジョブ開始] --> B[ChangelogStateBackend初期化]
    B --> C[StateChangelogWriter取得]
    C --> D[ステート操作]
    D --> E{変更タイプ?}
    E -->|ValueUpdate| F[valueUpdated()でログ記録]
    E -->|ValueAdd| G[valueAdded()でログ記録]
    E -->|ValueClear| H[valueCleared()でログ記録]
    F --> I[チェックポイント?]
    G --> I
    H --> I
    I -->|Yes| J[snapshot()]
    J --> K[persist()で差分永続化]
    K --> L[ChangelogStateBackendHandle作成]
    I -->|No| D
    L --> M{マテリアライゼーション?}
    M -->|Yes| N[initMaterialization()]
    N --> O[委譲先のスナップショット]
    O --> P[handleMaterializationResult()]
    P --> Q[Changelog truncation]
    M -->|No| D
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-36-01 | デリゲーションパターン | 委譲先StateBackendをラップ | 常時 |
| BR-36-02 | 再帰的委譲禁止 | DelegatingStateBackendの再帰的ラップ不可 | 初期化時 |
| BR-36-03 | CHECKPOINTタイプのみ | SavepointはネイティブSavepointを使用 | スナップショット時 |
| BR-36-04 | 連続マテリアライゼーション制限 | 前回マテリアライゼーション完了前は新規トリガー不可 | マテリアライゼーション時 |

### 計算ロジック

チェックポイント時のChangelogStateBackendHandle構成：
- getMaterializedSnapshot(): マテリアライズ済みスナップショット
- getNonMaterializedStateHandles(): 差分Changelogハンドル

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | RDBMSは使用しない |

### テーブル別操作詳細

該当なし（StateChangelogStorageを使用）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CHANGELOG | IOException | Changelogストレージ書き込み失敗 | ストレージ設定確認 |
| MATERIALIZE | Exception | マテリアライゼーション失敗 | 最大失敗回数超過で例外 |
| RESTORE | Exception | リストア失敗 | ステートハンドル確認 |

### リトライ仕様

- マテリアライゼーションは設定された最大失敗回数までリトライ
- Changelog永続化はチェックポイント全体のリトライに依存

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

- Changelogは順序付けられたシーケンス番号で管理
- チェックポイント確認時にConfirm、アボート時にReset

## パフォーマンス要件

- チェックポイント時間：差分のみ永続化により大幅短縮
- マテリアライゼーション：バックグラウンドで非同期実行

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

StateChangelogStorageへのアクセスには適切な権限設定が必要

## 備考

- @Internal（内部API）
- DelegatingStateBackendを実装
- 委譲先バックエンドの機能はそのまま利用可能

---

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

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

### 推奨読解順序

#### Step 1: 抽象基底クラスを理解する

AbstractChangelogStateBackendでデリゲーションパターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | AbstractChangelogStateBackend.java | `flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/` | 委譲パターン、createKeyedStateBackend() |

**読解のコツ**:
- **51行目**: DelegatingStateBackendを実装
- **66-76行目**: コンストラクタで委譲先バックエンドを保持、再帰的委譲を禁止
- **79-93行目**: createKeyedStateBackend()でrestore()を呼び出し

#### Step 2: ChangelogStateBackendを理解する

具体的なリストア処理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ChangelogStateBackend.java | `flink-statebackend-changelog/.../changelog/` | ConfigurableStateBackend、restore()実装 |

**読解のコツ**:
- **50-51行目**: AbstractChangelogStateBackendを継承、ConfigurableStateBackendを実装
- **59-70行目**: configure()で委譲先の設定を適用
- **74-141行目**: restore()でChangelogKeyedStateBackendを生成

#### Step 3: ChangelogKeyedStateBackendを理解する

ステート変更ログ記録の中心クラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ChangelogKeyedStateBackend.java | `flink-statebackend-changelog/.../changelog/` | スナップショット、マテリアライゼーション |

**読解のコツ**:
- **114-119行目**: 実装インターフェース（CheckpointableKeyedStateBackend、CheckpointListener、MaterializationTarget）
- **131行目**: keyedStateBackendが委譲先
- **145行目**: stateChangelogWriterでChangelog書き込み
- **399-445行目**: snapshot()でチェックポイント処理
- **853-905行目**: initMaterialization()でマテリアライゼーション開始
- **912-938行目**: handleMaterializationResult()で結果処理

#### Step 4: StateChangeLoggerを理解する

ステート変更のログ記録インターフェースを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StateChangeLogger.java | `flink-statebackend-changelog/.../changelog/` | ログ記録API定義 |
| 4-2 | KvStateChangeLoggerImpl.java | `flink-statebackend-changelog/.../changelog/` | Key-Valueステート用実装 |

**読解のコツ**:
- **StateChangeLogger 47-78行目**: valueUpdated、valueAdded、valueCleared等のAPI
- **KvStateChangeLoggerImpl 49-65行目**: コンストラクタでシリアライザ設定

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

```
ChangelogStateBackend
    │
    ├─ extends AbstractChangelogStateBackend
    │      ├─ implements DelegatingStateBackend
    │      └─ delegatedStateBackend（HashMap/RocksDB等）
    │
    ├─ restore()
    │      ├─ ChangelogBackendRestoreOperation.restore()
    │      ├─ ChangelogKeyedStateBackend
    │      │      ├─ keyedStateBackend（委譲先）
    │      │      ├─ stateChangelogWriter
    │      │      ├─ changelogStateFactory
    │      │      └─ changelogSnapshotState
    │      └─ PeriodicMaterializationManager
    │
    └─ ChangelogKeyedStateBackend
           ├─ createOrUpdateInternalState()
           │      └─ KvStateChangeLoggerImpl
           ├─ snapshot()
           │      └─ stateChangelogWriter.persist()
           ├─ initMaterialization()
           │      └─ keyedStateBackend.snapshot()
           └─ handleMaterializationResult()
                  └─ changelogTruncateHelper.materialized()
```

### データフロー図

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

ステート変更 ─────────────▶ ChangelogKeyedStateBackend ──▶ Changelog
                               │
                               ├─ 委譲先バックエンド
                               │      │
                               │      ▼
                               │  ステート実体（HashMap/RocksDB）
                               │
                               ├─ StateChangelogWriter
                               │      │
                               │      ▼
                               │  StateChangelogStorage
                               │      │
                               │      ▼
                               │  ChangelogStateHandle
                               │
                               └─ マテリアライゼーション
                                      │
                                      ▼
                               MaterializedSnapshot
                                      │
                                      ▼
                               ChangelogStateBackendHandle
                                 ├─ MaterializedSnapshot
                                 └─ ChangelogHandles
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AbstractChangelogStateBackend.java | `flink-statebackend-changelog/.../changelog/` | 抽象基底 | 委譲パターン |
| ChangelogStateBackend.java | `flink-statebackend-changelog/.../changelog/` | メインクラス | Configurable実装 |
| ChangelogKeyedStateBackend.java | `flink-statebackend-changelog/.../changelog/` | 実装 | キー付きステート管理 |
| StateChangeLogger.java | `flink-statebackend-changelog/.../changelog/` | インターフェース | ログ記録API |
| KvStateChangeLoggerImpl.java | `flink-statebackend-changelog/.../changelog/` | 実装 | KVステートログ記録 |
| ChangelogStateFactory.java | `flink-statebackend-changelog/.../changelog/` | ファクトリ | Changelogステート生成 |
| ChangelogTruncateHelper.java | `flink-statebackend-changelog/.../changelog/` | ヘルパー | ログ切り捨て管理 |
| PeriodicMaterializationManager.java | `flink-state-backends/.../common/` | マネージャ | 定期マテリアライゼーション |
| ChangelogBackendRestoreOperation.java | `flink-statebackend-changelog/.../restore/` | リストア | バックエンドリストア |
