# 機能設計書 26-チェックポイント

## 概要

本ドキュメントは、Apache Flinkのチェックポイント機能（flink-runtime モジュール内CheckpointCoordinator）の設計仕様を記述する。チェックポイント機能は、分散スナップショットによる耐障害性（Exactly-once保証）を提供する。

### 本機能の処理概要

CheckpointCoordinatorは、Flinkジョブの状態を一貫性のある分散スナップショットとして保存するための調整を行う。Chandy-Lamportアルゴリズムに基づく非同期バリアスナップショット方式を採用。

**業務上の目的・背景**：ストリーム処理において、障害発生時にデータ損失なく処理を再開できるようにする。Exactly-once処理保証を実現し、金融取引や請求処理など正確性が求められるユースケースに対応する。

**機能の利用シーン**：
- 定期的なチェックポイントの自動作成
- セーブポイント（手動スナップショット）の作成
- 障害発生時の状態復元
- ジョブアップグレード時の状態移行

**主要な処理内容**：
1. チェックポイントのトリガーと調整
2. チェックポイントバリアの送信
3. 各タスクからのACK収集
4. チェックポイント完了の確定
5. 古いチェックポイントのクリーンアップ

**関連システム・外部連携**：
- CheckpointStorage（状態保存先）
- TaskManager（タスク状態スナップショット）
- HighAvailabilityServices（メタデータ保存）

**権限による制御**：JobMaster経由でのみアクセス可能。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 10 | チェックポイント | 主機能 | チェックポイント統計・設定情報の取得 |

## 機能種別

状態管理 / 耐障害性 / データ保護

## 入力仕様

### 入力パラメータ

#### チェックポイントトリガー

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| checkpointType | CheckpointType | No | チェックポイント種別 | CHECKPOINT/SAVEPOINT |
| targetDirectory | String | No | 保存先ディレクトリ | セーブポイント時必須 |

#### チェックポイントACK

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| checkpointId | long | Yes | チェックポイントID | 正の整数 |
| executionAttemptID | ExecutionAttemptID | Yes | 実行試行ID | 有効なID |
| checkpointMetrics | CheckpointMetrics | Yes | メトリクス | null不可 |
| taskStateSnapshot | TaskStateSnapshot | No | タスク状態 | null許容 |

### 入力データソース

- SchedulerNGからのトリガー指示
- TaskExecutorからのACK/Decline

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| CompletedCheckpoint | CompletedCheckpoint | 完了チェックポイント情報 |
| CheckpointStatsSnapshot | CheckpointStatsSnapshot | 統計情報 |
| PendingCheckpoint | PendingCheckpoint | 進行中チェックポイント |

### 出力先

- CheckpointStorage（状態データ）
- CompletedCheckpointStore（メタデータ）
- REST API（統計情報）

## 処理フロー

### 処理シーケンス

```
1. チェックポイントトリガー
   └─ triggerCheckpoint() で開始
   └─ CheckpointStorageLocationを取得

2. バリア送信
   └─ 各ソースタスクにバリア送信
   └─ バリアがデータストリームを伝播

3. タスク状態スナップショット
   └─ 各タスクがバリア受信時に状態保存
   └─ 非同期でストレージに書き込み

4. ACK収集
   └─ 各タスクからACKを受信
   └─ 全タスクからACK受信で完了

5. チェックポイント完了
   └─ CompletedCheckpoint作成
   └─ CompletedCheckpointStoreに登録
   └─ 古いチェックポイントクリーンアップ
```

### フローチャート

```mermaid
flowchart TD
    A[triggerCheckpoint] --> B[PendingCheckpoint作成]
    B --> C[CheckpointStorageLocation取得]
    C --> D[ソースタスクにバリア送信]

    D --> E[バリア伝播]
    E --> F[タスク状態スナップショット]
    F --> G[ACK送信]

    G --> H{全タスクACK?}
    H -->|Yes| I[CompletedCheckpoint作成]
    H -->|No| J{タイムアウト?}
    J -->|Yes| K[チェックポイント失敗]
    J -->|No| H

    I --> L[CompletedCheckpointStore保存]
    L --> M[古いチェックポイントクリーンアップ]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 最小間隔 | minPauseBetweenCheckpoints以上の間隔を確保 | 連続トリガー時 |
| BR-002 | 同時実行数 | maxConcurrentCheckpointsを超えない | トリガー時 |
| BR-003 | タイムアウト | checkpointTimeout内に完了しない場合は失敗 | チェックポイント実行時 |
| BR-004 | 保持数 | 設定された数のチェックポイントを保持 | 完了時 |

### 計算ロジック

- チェックポイントIDの採番（単調増加）
- チェックポイントサイズの集計

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

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

本機能は直接データベースを操作しない。状態はCheckpointStorageに保存。

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CheckpointException | 処理エラー | チェックポイント失敗 | リトライ/スキップ |
| CheckpointDeclineException | ビジネスエラー | タスクがDecline | 原因調査 |
| CheckpointTimeoutException | タイムアウト | タイムアウト発生 | タイムアウト設定調整 |

### リトライ仕様

- チェックポイント失敗時: 次の間隔で再トリガー
- 連続失敗時: ジョブ失敗（設定による）

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

チェックポイントはアトミックに完了/失敗が決定。部分的な完了は発生しない。

## パフォーマンス要件

- チェックポイント完了: 設定されたタイムアウト以内
- 状態サイズ: ストレージ容量とスループットに依存

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

- チェックポイントデータの暗号化（オプション）
- ストレージアクセス権限の適切な設定

## 備考

- Unaligned Checkpointで高バックプレッシャー時のレイテンシ改善
- Incremental Checkpointで大規模状態の効率的な保存

---

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

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

### 推奨読解順序

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

チェックポイントで使用される主要なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PendingCheckpoint.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java` | 進行中チェックポイント |
| 1-2 | CompletedCheckpoint.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java` | 完了チェックポイント |
| 1-3 | CheckpointMetrics.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointMetrics.java` | メトリクス |

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

CheckpointCoordinatorの構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CheckpointCoordinator.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java` | メインクラス |

**主要処理フロー**:
1. **96-200行目**: フィールド定義（設定値、状態管理）
2. **triggerCheckpoint()**: チェックポイントトリガー
3. **acknowledgeCheckpoint()**: ACK受信処理
4. **completePendingCheckpoint()**: チェックポイント完了

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

```
CheckpointCoordinator
    │
    ├─ triggerCheckpoint()
    │      ├─ createPendingCheckpoint()
    │      ├─ checkpointStorageView.initializeLocationForCheckpoint()
    │      └─ triggerTasks()
    │
    ├─ acknowledgeCheckpoint()
    │      ├─ pendingCheckpoint.acknowledgeTask()
    │      └─ completePendingCheckpoint()
    │
    └─ completePendingCheckpoint()
           ├─ pendingCheckpoint.finalizeCheckpoint()
           ├─ completedCheckpointStore.addCheckpointAndSubsumeOldestOne()
           └─ checkpointsCleaner.cleanSubsumedCheckpoints()
```

### データフロー図

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

トリガー ─────────────▶ CheckpointCoordinator ────▶ CompletedCheckpoint
                              │
                              ▼
                    PendingCheckpoint
                              │
                              ▼
                    各TaskにBarrier送信
                              │
                              ▼
                    ACK収集（全タスク）
                              │
                              ▼
                    CompletedCheckpointStore
                              │
                              ▼
                    CheckpointStorage
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CheckpointCoordinator.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java` | ソース | チェックポイント調整 |
| PendingCheckpoint.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java` | ソース | 進行中チェックポイント |
| CompletedCheckpoint.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java` | ソース | 完了チェックポイント |
| CompletedCheckpointStore.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java` | ソース | チェックポイント保存 |
| CheckpointsCleaner.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointsCleaner.java` | ソース | クリーンアップ |
| CheckpointStorage.java | `flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorage.java` | ソース | ストレージインターフェース |
