# 通知設計書 24-jobManagerLostLeadership

## 概要

本ドキュメントは、Apache Flink の JobManager リーダーシップ喪失通知（jobManagerLostLeadership）の詳細設計を記載したものである。ジョブリーダーがリーダーシップを失った際に、タスクエグゼキュータに通知する仕組みを定義する。

### 本通知の処理概要

**業務上の目的・背景**：分散システムにおいて、リーダー変更は頻繁に発生する可能性がある。ジョブマネージャがリーダーシップを失った場合、タスクマネージャは既存の接続を切断し、新しいリーダーとの接続を待機する必要がある。本通知は、リーダーシップ喪失を迅速に検知し、適切なフェイルオーバー処理を行うための重要な通信メカニズムである。

**通知の送信タイミング**：`DefaultJobLeaderService` のリーダー取得リスナーがリーダーアドレスとして null または空文字列を受信した際に発生する。これは、既存のリーダーがリーダーシップを失い、まだ新しいリーダーが選出されていない状態を示す。

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

**通知内容の概要**：通知には、ジョブID と前のリーダーの JobMasterId が含まれる。これにより、タスクエグゼキュータはどのジョブに対するリーダーシップが失われたかを識別できる。

**期待されるアクション**：タスクエグゼキュータは、`disconnectJobManagerConnection()` メソッドを呼び出して、当該ジョブマスターとの接続を切断し、関連リソースのクリーンアップを行う。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（Executor 経由） |
| 優先度 | 高 |
| リトライ | なし |

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

`DefaultJobLeaderService` に登録された `JobLeaderListener` に対して通知が送信される。リーダー取得サービスから空のリーダー情報を受信した場合、現在のリーダーID（`currentJobMasterId`）を取得し、それを通知パラメータとして使用する。

## 通知テンプレート

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

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

### パラメータ

```java
void jobManagerLostLeadership(
    JobID jobId,
    JobMasterId jobMasterId
);
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| jobId | ジョブ識別子 | ジョブリーダーサービス | Yes |
| jobMasterId | 前のリーダーID | currentJobMasterId | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| リーダー喪失 | `notifyLeaderAddress(null/empty)` | 現在のリーダーIDが存在 | リーダー選出サービスからの通知 |
| 接続クローズ | RPC接続のクローズ時 | 現在のリーダーIDが存在 | 内部状態のクリーンアップ |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| リーダー未設定 | `currentJobMasterId` が null の場合 |
| サービス停止 | JobLeaderService が停止済みの場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[リーダー喪失検出] --> B[notifyLeaderAddress with null/empty]
    B --> C{サービス稼働中?}
    C -->|No| D[ログ出力して終了]
    C -->|Yes| E{現在のリーダー存在?}
    E -->|No| F[処理終了]
    E -->|Yes| G[jobManagerLostLeadership を Optional に保存]
    G --> H[closeRpcConnection]
    H --> I[ロック外でコールバック実行]
    I --> J[jobLeaderListener.jobManagerLostLeadership]
    J --> K[TaskExecutor.disconnectJobManagerConnection]
    K --> L[終了]
    F --> L
    D --> L
```

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

### 参照テーブル一覧

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

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

| データ構造 | クラス | 用途 | 備考 |
|-----------|------|------|------|
| currentJobMasterId | `JobMasterId` | 現在のリーダーID | JobManagerLeaderListener |
| rpcConnection | `RegisteredRpcConnection` | RPC接続管理 | JobManagerLeaderListener |

### 更新データ一覧

| データ構造 | 操作 | 概要 |
|-----------|------|------|
| currentJobMasterId | クリア | null に設定 |
| rpcConnection | クローズ | 接続の終了 |
| jobTable 接続 | 切断 | ジョブ接続情報の削除 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 接続クローズ失敗 | リソース解放エラー | ログ出力して続行 |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

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

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

- リーダーシップ喪失通知は内部コールバックであり、外部からのなりすましリスクは低い
- 既存接続のクリーンアップが適切に行われることを確認する必要がある

## 備考

- 本通知は高可用性モードで特に重要である
- `jobManagerGainedLeadership` と対になって動作する
- コールバックはロックの外で実行され、デッドロックを回避する設計となっている

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | JobMasterId.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | リーダーセッションIDの表現 |

**読解のコツ**: `JobMasterId` は UUID のラッパーであり、`fromUuidOrNull()` で null セーフに生成される。

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

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

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

**主要処理フロー**:
- **行 48-54**: `jobManagerLostLeadership()` メソッドシグネチャ
- パラメータは `jobId` と `jobMasterId` の2つのみ

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

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

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

**主要処理フロー**:
1. **行 294-337**: `notifyLeaderAddress()` - リーダー情報受信時の処理
2. **行 315-318**: リーダーアドレスが null/empty の場合の処理
3. **行 317**: `Optional.ofNullable(currentJobMasterId)` で通知対象を保存
4. **行 318**: `closeRpcConnection()` の呼び出し
5. **行 334-336**: ロック外での `jobManagerLostLeadership()` コールバック実行

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

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

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

**主要処理フロー**:
- **行 2547-2564**: `JobLeaderListenerImpl.jobManagerLostLeadership()` の実装
- **行 2549-2550**: ログ出力
- **行 2552-2563**: `runAsync()` で非同期に `disconnectJobManagerConnection()` を呼び出し
- **行 2557-2562**: 接続切断と例外メッセージの生成

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

```
LeaderRetrievalService (リーダー喪失検出)
    │
    ├─ JobManagerLeaderListener.notifyLeaderAddress(null/empty)  [行 294-337]
    │      │
    │      ├─ jobManagerLostLeadership を Optional に保存  [行 317]
    │      └─ closeRpcConnection()  [行 318, 357-364]
    │              ├─ rpcConnection.close()  [行 360]
    │              ├─ rpcConnection = null  [行 361]
    │              └─ currentJobMasterId = null  [行 362]
    │
    └─ コールバック実行（ロック外）  [行 334-336]
            └─ jobLeaderListener.jobManagerLostLeadership()
                    └─ TaskExecutor.JobLeaderListenerImpl  [行 2547-2564]
                            └─ runAsync()  [行 2552]
                                    └─ disconnectJobManagerConnection()  [行 2557]
```

### データフロー図

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

null/empty              ───▶  DefaultJobLeaderService   ───▶  コールバック通知
(リーダー喪失通知)             (接続クローズ)                  (TaskExecutor へ)
      │                              │                            │
      │                              ▼                            ▼
      │               closeRpcConnection()               JobLeaderListener
      │               (RPC接続終了)                       .jobManagerLostLeadership()
      │                              │                            │
      │                              ▼                            ▼
      │               currentJobMasterId = null        disconnectJobManagerConnection()
      │               (状態クリア)                      (接続切断)
      │                              │
      └──────────────────────────────┘
              前リーダーID
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| 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/` | ソース | タスクエグゼキュータ（受信側） |
| JobMasterId.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | ソース | リーダーセッションID |
| LeaderRetrievalService.java | `flink-runtime/src/main/java/org/apache/flink/runtime/leaderretrieval/` | ソース | リーダー取得サービス |
