# 機能設計書 91-ステートログ

## 概要

本ドキュメントは、Apache Flink の分散ステートログ（DSTL: Distributed State Transaction Log）機能について設計内容を記載する。DSTL は Changelog State Backend と連携し、ステート変更をファイルシステムに永続化することでチェックポイント処理を効率化する。

### 本機能の処理概要

DSTL は、Flink のステートバックエンドにおける変更（ステート変更）を、チェックポイントとは独立してファイルシステムに継続的に書き出す機能を提供する。これにより、チェックポイント時にはステート全体ではなく、前回のチェックポイント以降の差分のみを処理すれば良くなり、チェックポイント処理時間を大幅に短縮できる。

**業務上の目的・背景**：大規模なステートを持つストリーム処理ジョブでは、チェックポイント処理に長時間を要し、これがジョブのパフォーマンスや安定性に影響を与える。DSTL は、ステート変更を準リアルタイムで永続化することで、チェックポイント時のデータ転送量を最小化し、Exactly-Once セマンティクスを維持しながら高スループットを実現する。

**機能の利用シーン**：
- 大規模ステートを持つストリーム処理ジョブでチェックポイント時間を短縮したい場合
- 高頻度チェックポイントを実行したい場合
- ステート変更を継続的に永続化してリカバリ時間を短縮したい場合

**主要な処理内容**：
1. オペレーターからのステート変更をメモリに蓄積
2. 閾値（5MB デフォルト）を超えた場合にプリエンプティブにファイルシステムへアップロード
3. チェックポイント時に未アップロードのステート変更をバッチアップロード
4. シーケンス番号によるステート変更の順序管理
5. アップロード結果のハンドル管理とリカバリ時の読み込み

**関連システム・外部連携**：
- ファイルシステム（ローカル、HDFS、S3 等）への書き込み
- Changelog State Backend との連携
- チェックポイントコーディネーターとの連携

**権限による制御**：本機能に特定の権限制御はないが、ファイルシステムへの書き込み権限が必要。

## 関連画面

本機能は UI 画面と直接関連しない基盤機能である。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

データ永続化 / ステート管理 / チェックポイント最適化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| state.changelog.dstl.dfs.base-path | String | Yes | 変更ログファイルの保存先パス | 有効なファイルシステムパス |
| state.changelog.dstl.dfs.compression.enabled | Boolean | No | 圧縮の有効化（デフォルト: false） | - |
| state.changelog.dstl.dfs.preemptive-persist-threshold | MemorySize | No | プリエンプティブ永続化の閾値（デフォルト: 5MB） | 正の値 |
| state.changelog.dstl.dfs.batch.persist-delay | Duration | No | バッチ永続化の遅延（デフォルト: 10ms） | 非負の値 |
| state.changelog.dstl.dfs.batch.persist-size-threshold | MemorySize | No | バッチサイズ閾値（デフォルト: 10MB） | IN_FLIGHT_DATA_LIMIT 以下 |
| state.changelog.dstl.dfs.upload.buffer-size | MemorySize | No | アップロードバッファサイズ（デフォルト: 1MB） | 正の値 |
| state.changelog.dstl.dfs.upload.num-threads | Integer | No | アップロードスレッド数（デフォルト: 5） | 正の整数 |
| state.changelog.dstl.dfs.upload.max-in-flight | MemorySize | No | インフライトデータ上限（デフォルト: 100MB） | PERSIST_SIZE_THRESHOLD 以上 |
| state.changelog.dstl.dfs.upload.retry-policy | String | No | リトライポリシー（デフォルト: fixed） | none, fixed |
| state.changelog.dstl.dfs.upload.timeout | Duration | No | アップロードタイムアウト（デフォルト: 1秒） | 正の値 |
| state.changelog.dstl.dfs.upload.max-attempts | Integer | No | 最大リトライ回数（デフォルト: 3） | 正の整数 |

### 入力データソース

- オペレーターのステート変更（StateChange オブジェクト）
  - keyGroup: キーグループID
  - change: バイト配列形式の変更データ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ChangelogStateHandleStreamImpl | Object | アップロード結果のハンドル |
| UploadResult | Object | アップロード結果（シーケンス番号、オフセット、サイズ） |
| StreamStateHandle | Object | ファイルシステム上のステートハンドル |

### 出力先

- 設定されたファイルシステム（BASE_PATH で指定）
- ローカルリカバリが有効な場合はローカルディレクトリにも複製

## 処理フロー

### 処理シーケンス

```
1. FsStateChangelogStorage 初期化
   └─ JobID、設定、メトリクスグループを受け取り、アップローダーを生成

2. FsStateChangelogWriter 生成
   └─ オペレーターごとに固有の logId を割り当て

3. ステート変更の追加（append）
   └─ activeChangeSet にステート変更を蓄積
   └─ プリエンプティブ閾値超過時は即座に永続化

4. シーケンス番号の取得（nextSequenceNumber）
   └─ rollover() で現在の変更セットを notUploaded に移動
   └─ activeSequenceNumber をインクリメント

5. 永続化（persist）
   └─ 未アップロードの変更セットを抽出
   └─ StateChangeUploadScheduler にアップロードタスクを投入
   └─ 完了時にコールバックで結果を受け取り

6. アップロード完了処理
   └─ uploaded マップに結果を格納
   └─ SnapshotResult を生成して返却

7. トランケート（truncate）
   └─ マテリアライズされた変更を削除
   └─ lowestSequenceNumber を更新

8. クローズ（close）
   └─ アップローダーをクローズ
   └─ ローカルレジストリをクローズ
```

### フローチャート

```mermaid
flowchart TD
    A[ステート変更発生] --> B{activeChangeSetSize >= threshold?}
    B -->|Yes| C[プリエンプティブ永続化]
    B -->|No| D[activeChangeSet に追加]
    C --> E[StateChangeUploadScheduler.upload]
    D --> F{チェックポイント?}
    F -->|Yes| G[persist呼び出し]
    F -->|No| A
    G --> H[rollover - 変更セットを移動]
    H --> I[notUploaded から抽出]
    I --> J{toUpload が空?}
    J -->|Yes| K[uploaded から結果構築]
    J -->|No| L[UploadTask 生成]
    L --> E
    E --> M{アップロード成功?}
    M -->|Yes| N[handleUploadSuccess]
    M -->|No| O[handleUploadFailure]
    N --> P[uploaded に結果追加]
    O --> Q[highestFailed 更新]
    P --> R[SnapshotResult 返却]
    K --> R
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | シーケンス番号の一意性 | 各ステート変更セットは固有のシーケンス番号を持つ | 常時 |
| BR-02 | プリエンプティブ永続化 | activeChangeSetSize が閾値を超えたら自動的に永続化 | preemptive-persist-threshold 設定時 |
| BR-03 | インフライト制限 | インフライトデータ量が上限を超えるとバックプレッシャー | upload.max-in-flight 設定時 |
| BR-04 | リトライポリシー | アップロード失敗時は設定に従いリトライ | retry-policy が fixed の場合 |

### 計算ロジック

- インフライト容量計算: 現在アップロード中の全タスクのサイズ合計
- バッチサイズ計算: scheduledBytesCounter で追跡
- リトライ間隔: 失敗後 500ms（デフォルト）

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

本機能はデータベースを使用しない。ファイルシステムへの書き込みを行う。

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | ファイルシステムを使用 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | IO エラー | ファイルシステム書き込み失敗 | リトライポリシーに従いリトライ |
| IllegalArgumentException | バリデーションエラー | 無効なシーケンス番号範囲 | パラメータ確認 |
| CancellationException | キャンセル | クローズ時の未完了タスク | タスク失敗として処理 |

### リトライ仕様

- リトライポリシー: fixed（固定間隔）または none（リトライなし）
- 最大リトライ回数: デフォルト 3 回
- リトライ間隔: デフォルト 500ms
- タイムアウト: デフォルト 1 秒

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

- 各アップロードタスクは独立して実行
- 成功/失敗はコールバックで通知
- 失敗したシーケンス番号は highestFailed として記録
- チェックポイント確認時に uploaded から stopTracking を呼び出し

## パフォーマンス要件

- プリエンプティブ永続化により、チェックポイント時間を短縮
- バッチ処理により、小さな変更を効率的にまとめてアップロード
- インフライト制限により、メモリ使用量を制御
- 複数スレッドによる並列アップロード

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

- ファイルシステムへの書き込み権限が必要
- ステートデータはそのままファイルに書き込まれるため、必要に応じて暗号化を検討
- ローカルリカバリ使用時はローカルディスクへのアクセス権限が必要

## 備考

- DSTL は Experimental 機能として提供されている
- Changelog State Backend と組み合わせて使用
- ローカルリカバリとの併用が可能

---

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

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

### 推奨読解順序

#### Step 1: データ構造を理解する

まず、ステート変更がどのようなデータ構造で表現されるかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StateChangeSet.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeSet.java` | ステート変更セットの構造（logId, sequenceNumber, changes） |
| 1-2 | UploadResult.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/UploadResult.java` | アップロード結果の構造（sequenceNumber, offset, size, streamStateHandle） |
| 1-3 | FsStateChangelogOptions.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogOptions.java` | 設定オプションの定義 |

**読解のコツ**: StateChangeSet は logId（UUID）でバックエンドを識別し、sequenceNumber で順序を管理する。UploadResult はアップロード完了後の参照情報を保持する。

#### Step 2: エントリーポイントを理解する

処理の起点となるファクトリとストレージクラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | FsStateChangelogStorageFactory.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorageFactory.java` | SPI によるストレージ生成 |
| 2-2 | FsStateChangelogStorage.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java` | ストレージのメインクラス |

**主要処理フロー**:
1. **42-47行目**: IDENTIFIER として "filesystem" を定義
2. **50-57行目**: createStorage() でストレージインスタンスを生成
3. **60-63行目**: createStorageView() でリカバリ用ビューを生成

#### Step 3: Writer の動作を理解する

ステート変更の書き込み処理を担う Writer を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | FsStateChangelogWriter.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java` | 変更の蓄積と永続化 |

**主要処理フロー**:
- **171-191行目**: append() でステート変更を追加、preEmptiveFlushIfNeeded() で閾値チェック
- **210-218行目**: persist() でチェックポイント時の永続化開始
- **232-272行目**: persistInternal() でアップロードタスク生成と結果管理
- **360-371行目**: rollover() で activeChangeSet を notUploaded に移動

#### Step 4: アップロードスケジューラを理解する

バッチ処理とリトライを担うスケジューラを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StateChangeUploadScheduler.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeUploadScheduler.java` | スケジューラインターフェース |
| 4-2 | BatchingStateChangeUploadScheduler.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/BatchingStateChangeUploadScheduler.java` | バッチ処理実装 |

**主要処理フロー**:
- **158-188行目**: upload() でタスクをキューに追加、容量制限チェック
- **205-216行目**: scheduleUploadIfNeeded() で遅延またはサイズ閾値に基づくスケジュール
- **218-248行目**: drainAndSave() でキューから取り出してアップロード実行

#### Step 5: シリアライゼーション形式を理解する

ステート変更のファイル形式を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | StateChangeFormat.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/StateChangeFormat.java` | シリアライゼーション形式 |

**主要処理フロー**:
- **49-64行目**: write() で変更セットをソートしてシリアライズ
- **66-82行目**: writeChangeSet() でキーグループごとにグルーピング
- **95-149行目**: read() でイテレータ形式で読み込み

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

```
FsStateChangelogStorageFactory.createStorage()
    |
    +-- FsStateChangelogStorage
            |
            +-- createWriter()
            |       |
            |       +-- FsStateChangelogWriter
            |               |
            |               +-- append() / appendMeta()
            |               |       |
            |               |       +-- preEmptiveFlushIfNeeded()
            |               |               |
            |               |               +-- persistInternal()
            |               |
            |               +-- persist()
            |                       |
            |                       +-- persistInternal()
            |                               |
            |                               +-- StateChangeUploadScheduler.upload()
            |                                       |
            |                                       +-- BatchingStateChangeUploadScheduler
            |                                               |
            |                                               +-- drainAndSave()
            |                                                       |
            |                                                       +-- StateChangeUploader.upload()
            |                                                               |
            |                                                               +-- StateChangeFsUploader
            |
            +-- close()
```

### データフロー図

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

StateChange          FsStateChangelogWriter              ファイルシステム
(keyGroup, bytes) --> append() --> activeChangeSet
                           |
                           v
                      rollover()
                           |
                           v
                      notUploaded Map
                           |
                           v
                      persistInternal()
                           |
                           v
                  StateChangeUploadScheduler
                           |
                           v
                  BatchingStateChangeUploadScheduler
                           |
                           v
                  StateChangeFsUploader.upload() --> 変更ログファイル
                           |
                           v
                      UploadResult
                           |
                           v
                      uploaded Map --> ChangelogStateHandleStreamImpl
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| FsStateChangelogStorage.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | ストレージメインクラス |
| FsStateChangelogWriter.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | 変更書き込みクラス |
| FsStateChangelogStorageFactory.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | ファクトリクラス |
| FsStateChangelogStorageForRecovery.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | リカバリ用ストレージ |
| FsStateChangelogOptions.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | 設定オプション定義 |
| StateChangeUploadScheduler.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | スケジューラインターフェース |
| BatchingStateChangeUploadScheduler.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | バッチスケジューラ実装 |
| StateChangeUploader.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | アップローダーインターフェース |
| StateChangeFsUploader.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | ファイルシステムアップローダー |
| DuplicatingStateChangeFsUploader.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | ローカルリカバリ対応アップローダー |
| StateChangeSet.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | 変更セットデータ構造 |
| UploadResult.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | アップロード結果データ構造 |
| StateChangeFormat.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | シリアライゼーション形式 |
| TaskChangelogRegistry.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | タスクレジストリインターフェース |
| RetryPolicy.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | リトライポリシー |
| RetryingExecutor.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | リトライ付き実行器 |
| ChangelogStorageMetricGroup.java | `flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/` | ソース | メトリクスグループ |
| pom.xml | `flink-dstl/flink-dstl-dfs/` | 設定 | Maven ビルド設定 |
| org.apache.flink.runtime.state.changelog.StateChangelogStorageFactory | `flink-dstl/flink-dstl-dfs/src/main/resources/META-INF/services/` | 設定 | SPI 登録ファイル |
