# 通知設計書 23-jobManagerGainedLeadership

## 概要

本ドキュメントは、Apache Flink の JobManager リーダーシップ取得通知（jobManagerGainedLeadership）の詳細設計を記載したものである。ジョブマネージャがリーダーシップを取得し、タスクマネージャとの接続が確立された際に、タスクエグゼキュータに通知する仕組みを定義する。

### 本通知の処理概要

**業務上の目的・背景**：分散システムにおいて、Flink はリーダー選出メカニズムを用いてジョブマネージャの高可用性を実現している。タスクマネージャは、ジョブを実行するために正しいジョブマネージャ（リーダー）と接続する必要がある。本通知は、新しいリーダーが選出され接続が確立された際に、タスクエグゼキュータが適切なアクションを実行できるようにするための重要な通信メカニズムである。

**通知の送信タイミング**：`DefaultJobLeaderService` がジョブマネージャへの登録に成功し、`JMTMRegistrationSuccess` レスポンスを受信した際に発生する。具体的には、リーダー選出後のタスクマネージャ登録プロセスが正常に完了したタイミングである。

**通知の受信者**：`TaskExecutor` 内の `JobLeaderListenerImpl` が受信者となる。`JobLeaderService` を通じてタスクエグゼキュータに通知が配信される。

**通知内容の概要**：通知には、ジョブID、JobMasterGateway（RPC 通信用ゲートウェイ）、および登録成功メッセージ（`JMTMRegistrationSuccess`）が含まれる。これにより、タスクエグゼキュータはジョブマスターとの通信チャネルを確立できる。

**期待されるアクション**：タスクエグゼキュータは、`establishJobManagerConnection()` メソッドを呼び出して、ジョブマスターとの接続を確立し、スロットの割り当てやタスクの実行準備を行う。

## 通知種別

内部コールバック通知（Java インターフェースコールバック）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（Executor 経由） |
| 優先度 | 高 |
| リトライ | 登録リトライロジックに依存 |

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

`DefaultJobLeaderService` に登録された `JobLeaderListener` に対して通知が送信される。各ジョブに対してリーダー取得サービスが起動され、リーダーが検出されると登録プロセスが開始される。登録成功後に `onRegistrationSuccess()` コールバックが呼び出され、その中で `jobLeaderListener.jobManagerGainedLeadership()` が実行される。

## 通知テンプレート

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

| 項目 | 内容 |
|-----|------|
| インターフェース | `JobLeaderListener` |
| メソッド | `jobManagerGainedLeadership` |
| 形式 | Java インターフェースコールバック |

### パラメータ

```java
void jobManagerGainedLeadership(
    JobID jobId,
    JobMasterGateway jobManagerGateway,
    JMTMRegistrationSuccess registrationMessage
);
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| jobId | ジョブ識別子 | ジョブリーダーサービス | Yes |
| jobManagerGateway | RPC ゲートウェイ | 登録プロセス | Yes |
| registrationMessage | 登録成功メッセージ | ジョブマスターからのレスポンス | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 登録成功 | `onRegistrationSuccess()` | 登録が有効な場合 | ジョブマスターへの登録完了 |
| リーダー検出 | `notifyLeaderAddress()` | 新しいリーダー情報受信 | リーダー選出サービスからの通知 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 無効な登録 | リーダーセッションIDが現在のものと一致しない場合 |
| サービス停止 | JobLeaderService が停止済みの場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[リーダー情報受信] --> B[notifyLeaderAddress]
    B --> C{サービス稼働中?}
    C -->|No| D[ログ出力して終了]
    C -->|Yes| E{同一リーダー?}
    E -->|Yes| F[無視]
    E -->|No| G[既存接続クローズ]
    G --> H[新規RPC接続開始]
    H --> I[JobManagerRetryingRegistration]
    I --> J{登録成功?}
    J -->|No| K[リトライ/エラー処理]
    J -->|Yes| L[onRegistrationSuccess]
    L --> M{有効な登録?}
    M -->|No| N[ログ出力して終了]
    M -->|Yes| O[jobLeaderListener.jobManagerGainedLeadership]
    O --> P[TaskExecutor.establishJobManagerConnection]
    P --> Q[終了]
    F --> Q
    D --> Q
    N --> Q
    K --> I
```

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

### 参照テーブル一覧

該当なし（インメモリ管理）

### メモリ内データ構造

| データ構造 | クラス | 用途 | 備考 |
|-----------|------|------|------|
| jobLeaderServices | `Map<JobID, Tuple2<LeaderRetrievalService, JobManagerLeaderListener>>` | ジョブ別リーダーサービス管理 | DefaultJobLeaderService |
| rpcConnection | `RegisteredRpcConnection` | RPC接続管理 | JobManagerLeaderListener |
| currentJobMasterId | `JobMasterId` | 現在のリーダーID | JobManagerLeaderListener |

### 更新データ一覧

| データ構造 | 操作 | 概要 |
|-----------|------|------|
| jobTable | 更新 | ジョブ接続情報の確立 |
| ジョブマネージャ接続 | 確立 | 新規接続の確立 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 登録失敗 | ネットワーク障害等 | リトライ後、handleError 呼び出し |
| 無効なセッション | リーダー変更 | 処理スキップ |
| RPC タイムアウト | 通信タイムアウト | リトライ |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | RetryingRegistrationConfiguration に依存 |
| リトライ間隔 | 指数バックオフ |
| リトライ対象エラー | 一時的なネットワーク障害 |

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（システム稼働中常時）

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

- RPC 通信は Flink の認証・暗号化設定に従う
- リーダーセッションIDによる認証が行われる
- タスクマネージャは正当なジョブマスターにのみ接続する

## 備考

- 本通知は高可用性モードで特に重要である
- リーダー変更時には `jobManagerLostLeadership` と対になって動作する
- `JobLeaderListener` インターフェースの一部として定義されている

---

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

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

### 推奨読解順序

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

通知パラメータと関連データ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | JMTMRegistrationSuccess.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | 登録成功メッセージの構造 |
| 1-2 | JobMasterId.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | リーダーセッションIDの表現 |

**読解のコツ**: `JMTMRegistrationSuccess` にはリソースマネージャID や クラスタ情報が含まれる。

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

リスナーインターフェースと通知の起点を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | JobLeaderListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | リスナーインターフェース定義 |

**主要処理フロー**:
- **行 43-46**: `jobManagerGainedLeadership()` メソッドシグネチャ
- **行 28-32**: インターフェースの目的説明（Javadoc）

#### Step 3: 通知送信ロジックを理解する

DefaultJobLeaderService での通知生成を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | DefaultJobLeaderService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | ジョブリーダーサービス実装 |

**主要処理フロー**:
1. **行 294-337**: `notifyLeaderAddress()` - リーダー情報受信時の処理
2. **行 340-355**: `openRpcConnectionTo()` - RPC 接続の開始
3. **行 413-429**: `onRegistrationSuccess()` - 登録成功時のコールバック
4. **行 422-423**: `jobLeaderListener.jobManagerGainedLeadership()` の呼び出し

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

TaskExecutor での通知受信処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | タスクエグゼキュータ実装 |

**主要処理フロー**:
- **行 2529-2545**: `JobLeaderListenerImpl.jobManagerGainedLeadership()` の実装
- **行 2532-2544**: `runAsync()` で非同期に `establishJobManagerConnection()` を呼び出し

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

```
LeaderRetrievalService (リーダー検出)
    │
    ├─ JobManagerLeaderListener.notifyLeaderAddress()  [行 294-337]
    │      │
    │      ├─ closeRpcConnection()  [行 357-364]
    │      └─ openRpcConnectionTo()  [行 340-355]
    │              └─ JobManagerRegisteredRpcConnection.start()
    │                      └─ JobManagerRetryingRegistration
    │                              └─ gateway.registerTaskManager()  [RPC]
    │
    └─ JobManagerRegisteredRpcConnection.onRegistrationSuccess()  [行 413-429]
            │
            ├─ 有効性チェック  [行 415, 472-477]
            └─ jobLeaderListener.jobManagerGainedLeadership()  [行 422-423]
                    └─ TaskExecutor.JobLeaderListenerImpl
                            └─ runAsync()
                                    └─ establishJobManagerConnection()
```

### データフロー図

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

LeaderInformation      ───▶  DefaultJobLeaderService   ───▶  コールバック通知
(リーダーアドレス)             (登録・接続管理)               (TaskExecutor へ)
      │                              │                            │
      │                              ▼                            ▼
      │               JobManagerRetryingRegistration      JobLeaderListener
      │               (RPC登録プロセス)                    .jobManagerGainedLeadership()
      │                              │                            │
      │                              ▼                            ▼
      │               JMTMRegistrationSuccess           establishJobManagerConnection()
      │               (登録成功レスポンス)                (接続確立)
      │                              │
      └──────────────────────────────┘
              セッション情報
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| JobLeaderListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | ソース | リスナーインターフェース |
| DefaultJobLeaderService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | ソース | ジョブリーダーサービス実装 |
| TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/` | ソース | タスクエグゼキュータ（受信側） |
| JobMasterGateway.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | ソース | RPC ゲートウェイ |
| JMTMRegistrationSuccess.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | ソース | 登録成功メッセージ |
| LeaderRetrievalService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/` | ソース | リーダー取得サービス |
| RegisteredRpcConnection.java | `flink-runtime/src/main/java/org/apache/flink/runtime/registration/` | ソース | RPC 接続管理 |
