# 通知設計書 3-notifyCheckpointSubsumed

## 概要

本ドキュメントは、Apache Flinkにおけるチェックポイント置換通知（notifyCheckpointSubsumed）の設計を記載する。分散チェックポイントが後続のチェックポイントにより置き換えられた際に通知される内部専用のコールバック機能である。

### 本通知の処理概要

本通知は、Flinkの分散チェックポイント機構において、あるチェックポイントが後続のチェックポイントによって置換（subsume）された際に、InternalCheckpointListenerインターフェースを実装する内部コンポーネントに対して発行される通知である。

**業務上の目的・背景**：チェックポイントが他のチェックポイントによって包含された場合、古いチェックポイントに関連するリソースを早期にクリーンアップするために必要である。これにより、不要なリソースの保持を防ぎ、メモリやストレージの効率的な利用が可能となる。内部使用専用であり、ユーザーコードからは直接使用されない。

**通知の送信タイミング**：新しいチェックポイントが完了し、古いチェックポイントが不要になった時点で発行される。具体的には、チェックポイントIDがより大きなIDに包含された場合に通知される。

**通知の受信者**：InternalCheckpointListenerインターフェースを実装するFlink内部コンポーネントのみ。主にStateBackend、FileMergingSnapshotManager、ChangelogRegistry等の状態管理コンポーネントが対象。

**通知内容の概要**：置換されたチェックポイントのID（long型）が通知される。受信者はこのIDを使用して、対応するチェックポイントに関連するリソースのクリーンアップを実行する。

**期待されるアクション**：受信者は、通知されたチェックポイントIDに関連する補助的なリソース（一時ファイル、キャッシュ等）をクリーンアップする。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（タスク内で直接呼び出し） |
| 優先度 | 中 |
| リトライ | なし（ベストエフォート通知） |

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

InternalCheckpointListenerインターフェースを実装しているFlink内部コンポーネントに対して通知される。

## 通知テンプレート

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

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

### 本文テンプレート

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

### 添付ファイル

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

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| checkpointId | 置換されたチェックポイントの一意識別子 | SubtaskCheckpointCoordinatorImpl | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部イベント | チェックポイント完了通知処理 | より大きなIDが完了済み | 包含されたチェックポイントに対して通知 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| タスクがクローズ済み | タスクが終了している場合は通知をスキップ |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[新チェックポイント完了] --> B[SubtaskCheckpointCoordinatorImpl.notifyCheckpointComplete]
    B --> C{古いチェックポイントが存在?}
    C -->|Yes| D[NotifyCheckpointOperation.SUBSUME設定]
    D --> E[OperatorChain.notifyCheckpointSubsumed]
    E --> F[各内部コンポーネントに通知]
    F --> G[InternalCheckpointListener.notifyCheckpointSubsumed]
    C -->|No| H[終了]
```

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

### 参照テーブル一覧

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| Exception | リスナー内で例外発生 | タスク/ジョブの失敗とリカバリをトリガー |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0（リトライなし） |
| リトライ間隔 | - |
| リトライ対象エラー | なし |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | なし |
| 1日あたり上限 | なし |

### 配信時間帯

制限なし

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

- 内部通知のため、外部からのアクセスはない
- @Internalアノテーションにより、パブリックAPIではないことが明示されている

## 備考

- InternalCheckpointListenerはCheckpointListenerを継承した内部専用インターフェース
- ユーザーコードからは直接使用されない
- 主にリソースクリーンアップの最適化に使用される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | InternalCheckpointListener.java | `flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java` | インターフェース定義、@Internalアノテーションを確認 |

**読解のコツ**: CheckpointListenerを継承していること、@Internalアノテーションによりユーザー向けAPIではないことを理解する。

#### Step 2: 通知発行元を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SubtaskCheckpointCoordinatorImpl.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java` | notifyCheckpointComplete内のSUBSUME処理（481-482行目） |

#### Step 3: オペレータチェーンへの伝播を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | RegularOperatorChain.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/RegularOperatorChain.java` | notifyCheckpointSubsumedメソッド |
| 3-2 | StreamOperatorWrapper.java | `flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java` | notifyCheckpointSubsumedメソッド |

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

```
SubtaskCheckpointCoordinatorImpl.notifyCheckpointComplete()
    │
    └─ NotifyCheckpointOperation.SUBSUME判定
           │
           └─ OperatorChain.notifyCheckpointSubsumed(checkpointId)
                  │
                  └─ StreamOperatorWrapper.notifyCheckpointSubsumed(...)
                         │
                         └─ InternalCheckpointListener.notifyCheckpointSubsumed(checkpointId)
```

### データフロー図

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

チェックポイント完了 ───▶ SubtaskCheckpointCoordinator ───▶ OperatorChain
                               │                                │
                               │                                ▼
                               │                     InternalCheckpointListener
                               │                        (StateBackend等)
                               │                                │
                               │                                ▼
                               └─────────────────────▶ リソースクリーンアップ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| InternalCheckpointListener.java | `flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.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` | ソース | オペレータチェーンへの通知伝播 |
| FileMergingSnapshotManager.java | `flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManager.java` | ソース | 実装例：ファイルマージング用の通知処理 |
