# 通知設計書 31-notifyApplicationStatusChange

## 概要

本ドキュメントは、Apache Flinkにおけるアプリケーションステータス変更通知（notifyApplicationStatusChange）の設計仕様を記述したものである。

### 本通知の処理概要

notifyApplicationStatusChangeは、Flinkアプリケーションのライフサイクルステータスが変更された際に、登録されたリスナーに対して変更を通知する内部通知メカニズムである。この通知により、アプリケーションの状態変化を監視し、適切なリソース管理やアーカイブ処理を実行することが可能となる。

**業務上の目的・背景**：分散ストリーム処理フレームワークであるFlinkでは、アプリケーション（1つ以上のジョブを含む論理的な実行単位）のライフサイクル管理が重要である。アプリケーションがCREATED、RUNNING、FAILING、FAILED、CANCELING、CANCELED、FINISHEDといった状態を遷移する際に、Dispatcherなどの管理コンポーネントが適切な処理（アーカイブ、リソース解放など）を実行する必要がある。この通知機能により、状態遷移に応じた処理を疎結合で実装できる。

**通知の送信タイミング**：アプリケーションの状態遷移が発生した際に通知される。具体的には、AbstractApplicationクラスのtransitionState()メソッド内で、状態遷移が検証・実行された後、登録されているすべてのstatusListenersに対して順次通知が送信される。

**通知の受信者**：ApplicationStatusListenerインターフェースを実装し、AbstractApplication.registerStatusListener()メソッドで登録されたリスナーが受信者となる。典型的な受信者として、DispatcherクラスがApplicationStatusListenerを実装しており、アプリケーションのアーカイブ処理を担当する。

**通知内容の概要**：通知には、変更が発生したアプリケーションのID（ApplicationID）と、新しいステータス（ApplicationState）の2つの情報が含まれる。

**期待されるアクション**：受信者は、新しいステータスに応じて適切な処理を実行することが期待される。終端状態（FAILED、CANCELED、FINISHED）への遷移時には、Dispatcherがアプリケーションのアーカイブ処理を開始し、History Serverへのアーカイブや、内部ストアへの保存を行う。

## 通知種別

内部コールバック通知（Java インターフェースベースのリスナーパターン）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（リスナーのforEachで順次呼び出し） |
| 優先度 | 高（状態遷移に同期して即座に実行） |
| リトライ | なし |

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

AbstractApplication.statusListeners リストに登録されているすべてのApplicationStatusListenerインスタンスに対して、登録順に通知が送信される。リスナーの登録はregisterStatusListener()メソッドを通じて行われ、スレッドセーフではないため、同時実行環境での登録には注意が必要である。

## 通知テンプレート

### コールバック通知の場合

| 項目 | 内容 |
|-----|------|
| インターフェース | ApplicationStatusListener |
| メソッドシグネチャ | void notifyApplicationStatusChange(ApplicationID applicationId, ApplicationState newStatus) |

### パラメータ詳細

```java
/**
 * @param applicationId 状態変更が発生したアプリケーションのID
 * @param newStatus 遷移後の新しいアプリケーションステータス
 */
void notifyApplicationStatusChange(ApplicationID applicationId, ApplicationState newStatus);
```

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| applicationId | アプリケーション識別子 | AbstractApplication.applicationId | Yes |
| newStatus | 新しいステータス | transitionState()の引数 | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 状態遷移 | transitionToRunning() | CREATED -> RUNNING | アプリケーション実行開始時 |
| 状態遷移 | transitionToFailing() | RUNNING -> FAILING | 失敗検出時 |
| 状態遷移 | transitionToFailed() | FAILING -> FAILED | 失敗確定時 |
| 状態遷移 | transitionToCanceling() | CREATED/RUNNING -> CANCELING | キャンセル開始時 |
| 状態遷移 | transitionToCanceled() | CANCELING -> CANCELED | キャンセル完了時 |
| 状態遷移 | transitionToFinished() | RUNNING -> FINISHED | 正常完了時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 不正な状態遷移 | ALLOWED_TRANSITIONS で定義されていない遷移の場合、IllegalStateExceptionがスローされ通知は送信されない |
| リスナー未登録 | statusListenersが空の場合、forEachは何も実行しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[状態遷移メソッド呼び出し] --> B[transitionState内部メソッド]
    B --> C[validateTransition]
    C -->|検証失敗| D[IllegalStateException]
    C -->|検証成功| E[タイムスタンプ更新]
    E --> F[applicationState更新]
    F --> G[statusListeners.forEach]
    G --> H[各リスナーのnotifyApplicationStatusChange呼び出し]
    H --> I[終了]
    D --> J[例外スロー]
```

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

### 参照テーブル一覧

本通知はインメモリのリスナーパターンで実装されており、直接的なデータベースアクセスは行わない。ただし、通知を受信したDispatcherは以下の処理を行う：

| 操作 | 対象 | 用途 | 備考 |
|------|------|------|------|
| 参照 | partialExecutionGraphInfoStore | ジョブのExecutionGraphInfo取得 | 終端状態遷移時 |
| 参照 | applications Map | アプリケーション情報取得 | 終端状態遷移時 |

### 更新テーブル一覧

| 操作 | 対象 | 概要 |
|------|------|------|
| 削除 | applications Map | アーカイブ完了後にアプリケーションを削除 |
| 削除 | partialExecutionGraphInfoStore | アーカイブ完了後にジョブ情報を削除 |
| 追加 | archivedApplicationStore | アーカイブされたアプリケーションを保存 |
| 追加 | applicationArchivingFutures | アーカイブ処理のFutureを追跡 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| IllegalStateException | 不正な状態遷移が試みられた場合 | 例外がスローされ、状態遷移は中断される |
| リスナー内例外 | リスナー処理中に例外が発生した場合 | 後続のリスナーへの通知が中断される可能性がある |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（同期呼び出しのため） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（状態遷移発生時に即座に通知）

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

- 通知はJVMプロセス内の内部通信であり、ネットワーク経由では送信されない
- ApplicationIDは内部識別子であり、機密情報は含まれない
- リスナーの登録はスレッドセーフではないため、初期化フェーズで行う必要がある

## 備考

- Dispatcherは ApplicationStatusListener を実装しており、メインスレッドで実行される
- 終端状態への遷移時のアーカイブ処理は非同期で実行される（getMainThreadExecutor経由）
- History Serverへのアーカイブが失敗してもエラーはログ出力のみで処理は継続される

---

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

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

### 推奨読解順序

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

まず、通知に関連するデータ型とインターフェースを理解することが重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ApplicationStatusListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/application/ApplicationStatusListener.java` | 通知インターフェースの定義。notifyApplicationStatusChangeメソッドのシグネチャを確認 |
| 1-2 | ApplicationState.java | `flink-core/src/main/java/org/apache/flink/api/common/ApplicationState.java` | アプリケーションの状態列挙型。CREATED、RUNNING、FAILING、FAILED、CANCELING、CANCELED、FINISHEDの7状態と、終端状態の判定ロジック |
| 1-3 | ApplicationID.java | `flink-core/src/main/java/org/apache/flink/api/common/ApplicationID.java` | アプリケーション識別子 |

**読解のコツ**: ApplicationStateのisTerminalState()メソッドが、Dispatcherでのアーカイブ処理の分岐条件となる。

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

状態遷移の起点となるAbstractApplicationクラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AbstractApplication.java | `flink-runtime/src/main/java/org/apache/flink/runtime/application/AbstractApplication.java` | 状態遷移メソッド群とリスナー管理 |

**主要処理フロー**:
1. **76-81行目**: コンストラクタで初期状態CREATEDを設定
2. **148-150行目**: registerStatusListenerでリスナーを登録
3. **156-176行目**: ALLOWED_TRANSITIONSで許可される状態遷移を定義
4. **179-206行目**: 各状態への遷移メソッド（transitionToRunning等）
5. **208-220行目**: transitionStateで実際の状態遷移と通知を実行
6. **218-219行目**: statusListeners.forEachで全リスナーにnotifyApplicationStatusChangeを呼び出し

#### Step 3: 受信者側の処理を理解する

Dispatcherがどのように通知を処理するかを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Dispatcher.java | `flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java` | ApplicationStatusListenerの実装 |

**主要処理フロー**:
- **158行目**: DispatcherがApplicationStatusListenerを実装していることを確認
- **680-749行目**: notifyApplicationStatusChangeの実装
  - **682行目**: 終端状態かどうかを判定
  - **691-692行目**: アーカイブ開始のログ出力
  - **693-698行目**: アプリケーションの状態タイムスタンプを収集
  - **700-710行目**: 関連ジョブのExecutionGraphInfoを取得
  - **713-747行目**: 非同期でアーカイブ処理を実行

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

```
AbstractApplication.transitionState(ApplicationState targetState)
    │
    ├─ validateTransition(targetState)
    │      └─ ALLOWED_TRANSITIONS.get(applicationState)
    │
    ├─ statusTimestamps[targetState.ordinal()] = System.currentTimeMillis()
    │
    ├─ applicationState = targetState
    │
    └─ statusListeners.forEach(listener -> listener.notifyApplicationStatusChange(...))
           │
           └─ Dispatcher.notifyApplicationStatusChange(applicationId, newStatus)
                  │
                  ├─ [終端状態の場合]
                  │      │
                  │      ├─ applications.get(applicationId)
                  │      │
                  │      ├─ FutureUtils.combineAll(jobFutures)
                  │      │
                  │      └─ thenAcceptAsync(...)
                  │             │
                  │             ├─ ArchivedApplication作成
                  │             │
                  │             ├─ applications.remove(applicationId)
                  │             │
                  │             ├─ writeToArchivedApplicationStore(...)
                  │             │
                  │             └─ historyServerArchivist.archiveApplication(...)
                  │
                  └─ [非終端状態の場合] 何もしない
```

### データフロー図

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

状態遷移要求 ───▶ AbstractApplication.transitionState() ───▶ 状態更新完了
                        │
                        ▼
                  ApplicationStatusListener
                  .notifyApplicationStatusChange()
                        │
                        ▼
                  Dispatcher（終端状態の場合）
                        │
                        ├───▶ ArchivedApplication ───▶ archivedApplicationStore
                        │
                        └───▶ HistoryServerArchivist ───▶ History Server
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ApplicationStatusListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/application/ApplicationStatusListener.java` | インターフェース | 通知リスナーインターフェース定義 |
| AbstractApplication.java | `flink-runtime/src/main/java/org/apache/flink/runtime/application/AbstractApplication.java` | 抽象クラス | アプリケーションの基底クラス、状態遷移と通知送信を実装 |
| ApplicationState.java | `flink-core/src/main/java/org/apache/flink/api/common/ApplicationState.java` | 列挙型 | アプリケーション状態の定義 |
| Dispatcher.java | `flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java` | クラス | ApplicationStatusListenerの主要実装、アーカイブ処理を担当 |
| ArchivedApplication.java | `flink-runtime/src/main/java/org/apache/flink/runtime/application/ArchivedApplication.java` | クラス | アーカイブされたアプリケーションのデータ構造 |
| AbstractApplicationTest.java | `flink-runtime/src/test/java/org/apache/flink/runtime/application/AbstractApplicationTest.java` | テスト | 状態遷移と通知のテスト |
