# 通知設計書 1-notifyCheckpointComplete

## 概要

本ドキュメントは、Apache Flinkにおけるチェックポイント完了通知（notifyCheckpointComplete）の設計を記載する。分散チェックポイントが正常に完了しコミットされた際にリスナーに通知される重要なコールバック機能である。

### 本通知の処理概要

本通知は、Flinkの分散チェックポイント機構において、チェックポイントが正常に完了しコミットされた際に、CheckpointListenerインターフェースを実装するコンポーネントに対して発行される内部通知である。

**業務上の目的・背景**：Flinkのストリーム処理において、外部システムとのトランザクション連携（例：Kafkaへのオフセットコミット、データベースへのトランザクションコミット）を実現するために必要である。チェックポイントが完了した時点で、外部システムへの副作用を確定させることで、exactly-once semanticsを実現する。

**通知の送信タイミング**：CheckpointCoordinatorが全てのタスクからチェックポイント完了のACKを受信し、チェックポイントが正常にコミットされた後に発行される。具体的には、PendingCheckpointがCompletedCheckpointに昇格した後、各タスクに対して非同期で通知が送信される。

**通知の受信者**：CheckpointListenerインターフェースを実装する全てのコンポーネント。具体的には、StreamOperator、SourceReader、SplitEnumerator、OperatorCoordinator、およびユーザー定義関数（UDF）が対象となる。

**通知内容の概要**：完了したチェックポイントのID（long型）のみが通知される。受信者はこのIDを使用して、対応するチェックポイントに関連する処理（トランザクションコミット等）を実行する。

**期待されるアクション**：受信者は、通知されたチェックポイントIDに対応する保留中のトランザクションをコミットする。また、「チェックポイント包含契約」に従い、通知されたID以下の全ての保留中のチェックポイントを完了済みとして処理することが期待される。

## 通知種別

内部コールバック通知（プログラム内部でのメソッド呼び出し）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（RPC経由でタスクに送信後、タスク内で同期的に処理） |
| 優先度 | 高（チェックポイント完了後、できるだけ早く通知） |
| リトライ | なし（ベストエフォート通知） |

### 送信先決定ロジック

CheckpointListenerインターフェースを実装している全てのStreamOperatorおよびOperatorCoordinatorに対して、チェーン内の順序（末尾から先頭）で通知される。

## 通知テンプレート

### メソッドシグネチャ

```java
void notifyCheckpointComplete(long checkpointId) throws Exception;
```

### 本文テンプレート

```
（プログラム内部通知のため、テキストテンプレートは存在しない）
引数: checkpointId - 完了したチェックポイントのID（正の長整数）
```

### 添付ファイル

| ファイル名 | 形式 | 条件 | 説明 |
|----------|------|------|------|
| なし | - | - | 内部通知のため添付ファイルなし |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| checkpointId | 完了したチェックポイントの一意識別子 | CheckpointCoordinator.pendingCheckpoints | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部イベント | CheckpointCoordinator.completePendingCheckpoint | 全タスクからACK受信完了 | チェックポイントが正常に完了した場合 |
| 内部イベント | StopWithSavepoint完了 | Savepoint作成完了 | SavepointがSyncで完了した場合 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| タスクがクローズ済み | StreamOperatorWrapperがclosed=trueの場合、通知をスキップ |
| 既に通知済みのチェックポイント | latestReportCheckpointId以下のIDは通知をスキップ |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[CheckpointCoordinator: 全ACK受信] --> B[PendingCheckpoint → CompletedCheckpoint]
    B --> C[Execution.notifyCheckpointComplete呼び出し]
    C --> D[TaskManagerGateway経由でタスクに送信]
    D --> E[Task.notifyCheckpointCompleteAsync]
    E --> F[StreamTask.notifyCheckpointCompleteAsync]
    F --> G[SubtaskCheckpointCoordinatorImpl.notifyCheckpointComplete]
    G --> H[OperatorChain.notifyCheckpointComplete]
    H --> I{各オペレータに通知}
    I --> J[StreamOperator.notifyCheckpointComplete]
    J --> K[CheckpointListener.notifyCheckpointComplete]
```

## データベース参照・更新仕様

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| なし | - | 内部メモリ上の状態のみ使用 |

### テーブル別参照項目詳細

内部通知のためデータベース参照なし。

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| なし | - | 内部メモリ上の状態のみ更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| Exception | リスナー内で例外発生 | タスク/ジョブの失敗とリカバリをトリガー |
| タイムアウト | 通知処理が長時間完了しない | タスクマネージャーによる強制終了 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0（リトライなし） |
| リトライ間隔 | - |
| リトライ対象エラー | なし（ベストエフォート通知） |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | なし（チェックポイント間隔に依存） |
| 1日あたり上限 | なし |

### 配信時間帯

制限なし（チェックポイント完了時に即座に通知）

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

- 内部通知のため、外部からのアクセスはない
- チェックポイントIDは順序保証のために使用され、機密情報は含まない
- リスナー実装内でのブロッキング処理は非推奨（パフォーマンス影響）

## 備考

- 「チェックポイント包含契約」：より大きなIDのチェックポイントが完了した場合、それより小さいIDの全てのチェックポイントも完了したとみなす
- 通知は「ベストエフォート」であり、障害/リストア直後には通知がスキップされる可能性がある
- 実装者は、通知が来なかった場合も正しく動作するよう設計する必要がある

---

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

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

### 推奨読解順序

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

まず、CheckpointListenerインターフェースの定義を理解することが重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CheckpointListener.java | `flink-core/src/main/java/org/apache/flink/api/common/state/CheckpointListener.java` | インターフェース定義、JavaDocのチェックポイント包含契約を理解 |

**読解のコツ**: JavaDocに記載されている「Checkpoint Subsuming Contract」は、実装者が必ず理解すべき重要な概念である。

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

処理の起点となるCheckpointCoordinatorを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CheckpointCoordinator.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java` | チェックポイント完了時の通知発行ロジック |

**主要処理フロー**:
1. **completePendingCheckpoint()**: PendingCheckpointをCompletedCheckpointに変換
2. **sendAcknowledgeMessages()**: 各タスクへの通知送信をトリガー

#### Step 3: タスクへの通知伝播を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamTask.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java` | notifyCheckpointCompleteAsync()とnotifyCheckpointComplete()メソッド（1528-1595行目） |
| 3-2 | SubtaskCheckpointCoordinatorImpl.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java` | notifyCheckpointComplete()メソッド（408行目以降） |
| 3-3 | RegularOperatorChain.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/RegularOperatorChain.java` | notifyCheckpointComplete()メソッド（141-153行目） |

**主要処理フロー**:
- **StreamTask.java 1528行**: notifyCheckpointCompleteAsync - 非同期通知のエントリーポイント
- **StreamTask.java 1580行**: notifyCheckpointComplete - 実際の通知処理
- **StreamTask.java 1589行**: SubtaskCheckpointCoordinatorへの委譲

#### Step 4: オペレータへの通知を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | StreamOperatorWrapper.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java` | notifyCheckpointComplete()メソッド（102-106行目） |
| 4-2 | StreamOperator.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java` | CheckpointListenerを継承していることを確認（47行目） |
| 4-3 | AbstractStreamOperator.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java` | デフォルト実装の確認 |

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

```
CheckpointCoordinator.completePendingCheckpoint()
    │
    ├─ Execution.notifyCheckpointComplete(checkpointId)
    │      │
    │      └─ TaskManagerGateway.notifyCheckpointComplete(...)
    │             │
    │             └─ TaskExecutor.confirmCheckpoint(...)
    │                    │
    │                    └─ Task.notifyCheckpointComplete(checkpointId)
    │                           │
    │                           └─ StreamTask.notifyCheckpointCompleteAsync(checkpointId)
    │                                  │
    │                                  └─ StreamTask.notifyCheckpointComplete(checkpointId)
    │                                         │
    │                                         └─ SubtaskCheckpointCoordinatorImpl.notifyCheckpointComplete(...)
    │                                                │
    │                                                └─ OperatorChain.notifyCheckpointComplete(checkpointId)
    │                                                       │
    │                                                       └─ StreamOperatorWrapper.notifyCheckpointComplete(...)
    │                                                              │
    │                                                              └─ StreamOperator.notifyCheckpointComplete(checkpointId)
    │
    └─ OperatorCoordinatorHolder.notifyCheckpointComplete(checkpointId)
```

### データフロー図

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

CheckpointCoordinator ───▶ TaskManagerGateway(RPC) ───▶ Task
     │                                                     │
     │                                                     ▼
     │                                               StreamTask
     │                                                     │
     │                                                     ▼
     │                                         SubtaskCheckpointCoordinator
     │                                                     │
     │                                                     ▼
     │                                               OperatorChain
     │                                                     │
     │                                                     ▼
     └──────────────────────────────────────────▶ CheckpointListener
                                                  (各オペレータ/UDF)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CheckpointListener.java | `flink-core/src/main/java/org/apache/flink/api/common/state/CheckpointListener.java` | インターフェース | 通知を受け取るリスナーの契約定義 |
| CheckpointCoordinator.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java` | ソース | チェックポイント全体を管理し、完了時に通知を発行 |
| StreamTask.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java` | ソース | タスク側での通知受信と処理 |
| SubtaskCheckpointCoordinatorImpl.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java` | ソース | サブタスクレベルでのチェックポイント通知管理 |
| RegularOperatorChain.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/RegularOperatorChain.java` | ソース | オペレータチェーンへの通知伝播 |
| StreamOperatorWrapper.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java` | ソース | 個別オペレータへの通知委譲 |
| StreamOperator.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java` | インターフェース | オペレータが実装すべきインターフェース |
| AbstractStreamOperator.java | `flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java` | ソース | オペレータの基底クラス |
