# 通知設計書 14-SparkListenerExecutorRemoved

## 概要

本ドキュメントは、Apache SparkのSparkListenerExecutorRemovedイベント通知の設計仕様を記載する。このイベントはドライバがExecutorを削除した際に発火し、LiveListenerBus経由で非同期に配信される。

### 本通知の処理概要

SparkListenerExecutorRemovedは、クラスタからExecutorが削除された際にリスナーバスへ通知を行うイベントである。

**業務上の目的・背景**：Sparkクラスタでは、Executorの障害、動的リソース管理によるスケールダウン、デコミッション、ユーザによる明示的なkill等の理由でExecutorが削除される。この通知により、Spark UIからのExecutor情報の更新（非活性化表示）、イベントログへの記録、HeartbeatReceiverからの監視解除、動的リソース管理での追跡情報更新が可能となる。

**通知の送信タイミング**：CoarseGrainedSchedulerBackendのremoveExecutor処理内で`listenerBus.post(SparkListenerExecutorRemoved(...))`が呼び出される。主に487行目（正常な削除）と502行目（既にメタデータが削除済みの場合）の2箇所で発火する。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装クラス。主要な受信者はAppStatusListener（UI更新）、EventLoggingListener（イベントログ）、HeartbeatReceiver（監視解除）、ExecutorMonitor（追跡更新）である。

**通知内容の概要**：Executor削除の時刻、ExecutorのID、削除理由（文字列）が含まれる。

**期待されるアクション**：受信者はExecutorの非活性化をUIに反映し、関連するRDD分散情報やストレージ情報のクリーンアップを実行する。HeartbeatReceiverは当該Executorの監視を停止する。

## 通知種別

アプリ内イベント通知（SparkListenerEvent）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、AsyncEventQueue使用） |
| 優先度 | 中（通常のイベントキュー処理） |
| リトライ | なし |

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

LiveListenerBusに登録された全てのAsyncEventQueueに対してブロードキャスト配信される。SparkListenerBus.doPostEventメソッド（64行目）で`listener.onExecutorRemoved(executorRemoved)`が呼び出される。

## 通知テンプレート

### メール通知の場合

該当なし。本通知はSpark内部のイベントバスを通じたプログラム間通知である。

### 本文テンプレート

```
イベント種別: SparkListenerExecutorRemoved
JSON形式:
{
  "Event": "SparkListenerExecutorRemoved",
  "Timestamp": <timestamp>,
  "Executor ID": "<executorId>",
  "Removed Reason": "<reason>"
}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | Executor削除時刻（ミリ秒） | System.currentTimeMillis() | Yes |
| executorId | ExecutorのID文字列 | CoarseGrainedSchedulerBackend内部処理 | Yes |
| reason | 削除理由の文字列 | lossReason.toString（ExecutorKilled, ExecutorDecommission等） | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | Executor削除（正常） | executorDataMapにExecutorが存在する場合 | CoarseGrainedSchedulerBackend 487行目 |
| 内部処理 | Executor削除（メタデータなし） | executorDataMapにExecutorが存在しない場合 | CoarseGrainedSchedulerBackend 502行目（SPARK-35011対応） |
| 内部処理 | Executorのkill | 動的リソース管理からのkill要求 | ExecutorAllocationManager経由 |
| 内部処理 | Executorのデコミッション完了 | デコミッション処理完了時 | デコミッション完了時にExecutor削除が連鎖 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| LiveListenerBus停止済み | stop()後はイベントがドロップされる |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[Executor削除要求] --> B[CoarseGrainedSchedulerBackend.DriverEndpoint.removeExecutor]
    B --> C{executorDataMapに存在?}
    C -->|Yes| D[ExecutorData削除・lossReason決定]
    D --> E[scheduler.executorLost呼び出し]
    E --> F[listenerBus.post SparkListenerExecutorRemoved 487行目]
    C -->|No| G[BlockManager非同期削除]
    G --> H[listenerBus.post SparkListenerExecutorRemoved 502行目]
    F --> I[LiveListenerBus.post]
    H --> I
    I --> J[各AsyncEventQueueにイベント配信]
    J --> K[SparkListenerBus.doPostEvent]
    K --> L{受信リスナー}
    L -->|AppStatusListener| M[Executor非活性化・RDD情報更新]
    L -->|EventLoggingListener| N[イベントログ書き込み]
    L -->|HeartbeatReceiver| O[ハートビート監視解除]
    L -->|ExecutorMonitor| P[Executor追跡情報削除]
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| executorDataMap（インメモリMap） | 削除対象Executorの情報参照 | CoarseGrainedSchedulerBackend内部 |

### テーブル別参照項目詳細

#### executorDataMap

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| executorId | 削除対象の特定 | 引数のexecutorIdで検索 |
| ExecutorData | 削除理由の決定に使用 | executorDataMapから取得 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| executorDataMap（インメモリMap） | DELETE | 対象ExecutorDataの削除 |
| KVStore（AppStatusListener経由） | UPDATE | Executor情報の非活性化更新 |

#### 送信ログテーブル

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| INSERT | Event | SparkListenerExecutorRemoved | イベントログに書き込み |
| INSERT | Timestamp | System.currentTimeMillis() | 削除時刻 |
| INSERT | Executor ID | executorId | ExecutorのID |
| INSERT | Removed Reason | reason | 削除理由 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| キュー満杯 | AsyncEventQueueのキャパシティ超過 | イベントがドロップされる |
| リスナー例外 | リスナーのonExecutorRemovedで例外発生 | ListenerBusが例外をキャッチしログに記録 |
| メタデータ不在 | executorDataMapにExecutorが存在しない | 502行目のパスでイベントを発火し、BlockManagerの非同期削除も実行 |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし。

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

イベントにはExecutorIDと削除理由文字列が含まれる。削除理由にはデコミッションメッセージやkillの理由が含まれる場合があるが、機密情報は通常含まれない。Spark UIのACLとイベントログのパーミッションによるアクセス制御で十分である。

## 備考

- SparkListenerExecutorRemovedはSparkListenerExecutorAddedと対になるイベントである。
- AppStatusListenerでは、削除されたExecutorに関連するRDDの分散情報やパーティション情報もクリーンアップされる（235-281行目）。
- SPARK-35011対応として、executorDataMapにメタデータがない場合でもイベントを発火し、UIの不整合を防止している。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 123行目: case class SparkListenerExecutorRemoved(time, executorId, reason)の定義 |

**読解のコツ**: reasonはString型で、ExecutorLossReason.toStringの結果が設定される。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CoarseGrainedSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala` | 475-506行目: removeExecutor処理。executorDataMapの有無で2つのパスに分岐 |

**主要処理フロー**:
1. **475-483行目**: lossReasonの決定（ExecutorKilled, ExecutorDecommission等）
2. **484行目**: totalCoreCountの減算
3. **486行目**: scheduler.executorLost呼び出し
4. **487-488行目**: `listenerBus.post(SparkListenerExecutorRemoved(...))` -- メインパス
5. **495行目**: BlockManagerの非同期削除（メタデータ不在時）
6. **502-503行目**: `listenerBus.post(SparkListenerExecutorRemoved(...))` -- フォールバックパス

#### Step 3: 受信リスナーの処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | 235-282行目: onExecutorRemovedでExecutor非活性化とRDD情報のクリーンアップ |
| 3-2 | HeartbeatReceiver.scala | `core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala` | 206行目: onExecutorRemovedでハートビート監視解除 |
| 3-3 | ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | 355行目: onExecutorRemovedでExecutor追跡情報削除 |
| 3-4 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 189行目: イベントログに記録 |

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

```
Executor削除要求
    |
    +-- CoarseGrainedSchedulerBackend.DriverEndpoint.removeExecutor()
            |
            +-- executorDataMap.remove(executorId)
            +-- scheduler.executorLost(executorId, lossReason)
            +-- listenerBus.post(SparkListenerExecutorRemoved)
                    |
                    +-- LiveListenerBus.post(event)
                            |
                            +-- postToQueues(event)
                                    |
                                    +-- SparkListenerBus.doPostEvent()
                                            |
                                            +-- AppStatusListener.onExecutorRemoved()
                                            |       +-- Executor非活性化
                                            |       +-- RDD分散情報クリーンアップ
                                            +-- HeartbeatReceiver.onExecutorRemoved()
                                            +-- ExecutorMonitor.onExecutorRemoved()
                                            +-- EventLoggingListener.onExecutorRemoved()
```

### データフロー図

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

Executor削除要求 -------> CoarseGrainedSchedulerBackend -------> SparkListenerExecutorRemoved
(executorId,               .removeExecutor()                       イベント
 reason)                        |
                                v
                          LiveListenerBus.post() ---------------> AsyncEventQueue
                                                                       |
                                                                       v
                                                                 AppStatusListener --> KVStore更新
                                                                 HeartbeatReceiver --> 監視解除
                                                                 ExecutorMonitor --> 追跡削除
                                                                 EventLoggingListener --> ログ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントcase classの定義 |
| CoarseGrainedSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala` | ソース | イベントの発火元 |
| LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | ソース | 非同期イベントバス |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | UI用データ更新・RDDクリーンアップ |
| HeartbeatReceiver.scala | `core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala` | ソース | ハートビート監視解除 |
| ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | ソース | 動的リソース管理 |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログ記録 |
| JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | ソース | JSON変換 |
