# 通知設計書 10-notifyLeaderAddress

## 概要

本ドキュメントは、Apache Flinkにおけるリーダーアドレス通知（notifyLeaderAddress）の設計を記載する。新しいリーダーが選出された際にリスナーに通知されるコールバック機能である。

### 本通知の処理概要

本通知は、Flinkの高可用性（HA）機構において、リーダー取得サービス（LeaderRetrievalService）がLeaderRetrievalListenerインターフェースを実装するコンポーネントに対して発行するコールバックである。

**業務上の目的・背景**：分散システムにおいて、クライアントやワーカーノードは現在のリーダーの場所を知る必要がある。このコールバックにより、TaskManager、クライアント、その他のコンポーネントは、現在有効なリーダー（JobManager、ResourceManager等）のアドレスを取得し、通信を確立できる。

**通知の送信タイミング**：LeaderRetrievalDriverがHAバックエンド（ZooKeeper、Kubernetes等）からリーダー情報の変更を検出した際に発行される。リーダーの選出、リーダーの変更、またはリーダーシップの喪失時に通知される。

**通知の受信者**：LeaderRetrievalListenerインターフェースを実装し、LeaderRetrievalServiceに登録されたコンポーネント。具体的には、TaskExecutor、JobMaster、DefaultJobLeaderIdService、DefaultJobLeaderServiceなどが対象。

**通知内容の概要**：新しいリーダーのアドレス（String型、nullableな）とリーダーセッションID（UUID型、nullable）が渡される。両方がnullの場合、リーダーシップが失われたことを示す。

**期待されるアクション**：受信者は、新しいリーダーアドレスを使用して通信先を更新する。リーダーシップが失われた場合（両引数がnull）は、適切な待機またはエラー処理を行う。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期 |
| 優先度 | 高（リーダー変更は重要イベント） |
| リトライ | なし |

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

LeaderRetrievalServiceに登録されたLeaderRetrievalListenerに対して通知される。

## 通知テンプレート

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

```java
void notifyLeaderAddress(@Nullable String leaderAddress, @Nullable UUID leaderSessionID);
```

### 本文テンプレート

```
（プログラム内部通知のため、テキストテンプレートは存在しない）
引数:
  - leaderAddress: リーダーのアドレス（nullの場合はリーダーシップ喪失を示す）
  - leaderSessionID: リーダーセッションの一意識別子（nullの場合はリーダーシップ喪失を示す）
```

### 添付ファイル

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

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| leaderAddress | リーダーの通信アドレス | HAバックエンド（ZK/K8s等） | 条件付き |
| leaderSessionID | リーダーセッションの一意識別子 | HAバックエンド | 条件付き |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部イベント | LeaderRetrievalDriver.notifyLeaderAddress | リーダー情報変更検出 | 新しいリーダー選出時 |
| 内部イベント | リーダーシップ喪失 | 両引数がnull | リーダーが存在しない状態 |
| 内部イベント | リーダー変更 | 新しいアドレス/セッションID | リーダーの切り替え時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 同一リーダー情報 | 前回と同じアドレス/セッションIDの場合はスキップ |
| サービス停止中 | LeaderRetrievalServiceがrunning=falseの場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[HAバックエンドでリーダー情報変更] --> B[LeaderRetrievalDriver.notifyLeaderAddress]
    B --> C[DefaultLeaderRetrievalService.notifyLeaderAddress]
    C --> D{前回と同じ情報?}
    D -->|Yes| E[スキップ]
    D -->|No| F[lastLeaderAddress/lastLeaderSessionID更新]
    F --> G[LeaderRetrievalListener.notifyLeaderAddress呼び出し]
    G --> H[リスナーがリーダーアドレス更新]
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| なし | - | ZooKeeper/Kubernetes等の外部ストレージを参照 |

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| Exception | リスナー内で例外発生 | handleErrorコールバックで通知 |
| 接続喪失 | HAバックエンドとの接続切断 | handleError経由で通知 |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし

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

- 内部通知のため、外部からのアクセスはない
- リーダーアドレスは通常、内部ネットワークアドレスを含む
- セッションIDは認証トークンではなく、リーダーシップの有効性検証に使用される

## 備考

- 両引数がnullの場合、リーダーシップが失われたことを示す（新しいリーダーがまだ選出されていない）
- handleErrorメソッドでエラー通知を受け取る
- LeaderContender（リーダー選出参加側）とは異なり、リーダー情報の購読側のインターフェース
- 同じリーダー情報の場合は通知がスキップされる（重複通知防止）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | LeaderRetrievalListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalListener.java` | インターフェース定義（26-50行目） |

**読解のコツ**: 両引数がnullの場合の意味（リーダーシップ喪失）を理解する。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DefaultLeaderRetrievalService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalService.java` | notifyLeaderAddressメソッド（124-159行目） |

**主要処理フロー**:
- **行126-127**: LeaderInformationからアドレスとセッションIDを抽出
- **行131-132**: 前回と同じ情報かチェック
- **行145-146**: lastLeaderAddress/lastLeaderSessionIDを更新
- **行149**: リスナーに通知

#### Step 3: 利用例を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java` | リスナー実装例 |
| 3-2 | DefaultJobLeaderService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobLeaderService.java` | リスナー実装例 |

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

```
HAバックエンド（ZK/K8s等）
    │
    └─ LeaderRetrievalDriver.notifyLeaderAddress(LeaderInformation)
           │
           └─ DefaultLeaderRetrievalService.notifyLeaderAddress(LeaderInformation)
                  │
                  └─ LeaderRetrievalListener.notifyLeaderAddress(address, sessionID)
                         │
                         ├─ TaskExecutor
                         ├─ JobMaster
                         ├─ DefaultJobLeaderIdService
                         └─ DefaultJobLeaderService
```

### データフロー図

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

HAバックエンド ───▶ LeaderRetrievalDriver ───▶ DefaultLeaderRetrievalService
(ZooKeeper/K8s)                                     │
                                                    ▼
                                            重複チェック
                                                    │
                                                    ▼
                                         LeaderRetrievalListener
                                        (アドレス/セッションID更新)
                                                    │
                                                    ▼
                                           リーダーとの通信確立
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| LeaderRetrievalListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalListener.java` | インターフェース | リーダー情報購読者の契約定義 |
| DefaultLeaderRetrievalService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalService.java` | ソース | リーダー取得サービスの実装 |
| LeaderRetrievalDriver.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/LeaderRetrievalDriver.java` | インターフェース | バックエンドドライバーの契約 |
| TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java` | ソース | 実装例：TaskManager |
| DefaultJobLeaderService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobLeaderService.java` | ソース | 実装例：ジョブリーダーサービス |
| StandaloneLeaderRetrievalService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/StandaloneLeaderRetrievalService.java` | ソース | 非HAモード用の実装 |
