# 通知設計書 11-SparkListenerBlockManagerRemoved

## 概要

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

### 本通知の処理概要

SparkListenerBlockManagerRemovedは、クラスタ内のBlockManagerが削除された際にストレージ管理層からスケジューラのリスナーバスへ通知を行うイベントである。

**業務上の目的・背景**：Sparkクラスタでは各ExecutorがBlockManagerを保持し、RDDパーティションやブロードキャスト変数などのデータブロックを管理している。Executorの終了やクラスタからの離脱、あるいは管理上の削除操作が行われた場合、対応するBlockManagerも削除される。この通知により、Spark UIの更新、イベントログへの記録、アプリケーションステータスの更新が適切に行われ、クラスタのストレージリソース状態をリアルタイムに把握することが可能となる。

**通知の送信タイミング**：BlockManagerMasterEndpointの`removeBlockManager`メソッド内で、BlockManagerの登録情報がクリアされた後に`listenerBus.post(SparkListenerBlockManagerRemoved(...))`が呼び出される。具体的には、Executorのハートビートタイムアウト、Executorの明示的な削除、Executorのデコミッション完了時などに発火する。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装クラスが受信する。主要な受信者はAppStatusListener（Spark UIの更新）、EventLoggingListener（イベントログ記録）、およびユーザ定義のカスタムリスナーである。

**通知内容の概要**：削除されたBlockManagerのID（BlockManagerId、ホスト・ポート・executorIdを含む）と、削除が発生した時刻（Unixタイムスタンプ）が含まれる。

**期待されるアクション**：受信者はBlockManagerの削除をUIに反映し、関連するストレージ情報のクリーンアップ処理を実行する。また、イベントログに記録してHistory Serverでの履歴参照を可能にする。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、AsyncEventQueue使用） |
| 優先度 | 中（通常のイベントキュー処理） |
| リトライ | なし（イベントキューが満杯の場合はドロップされる可能性あり） |

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

LiveListenerBusに登録された全てのAsyncEventQueueに対してブロードキャスト配信される。各キュー（shared, appStatus, eventLog, executorManagement）に登録されたリスナーがそれぞれ受信する。SparkListenerBus.doPostEventメソッドでイベントタイプに基づくパターンマッチングが行われ、`listener.onBlockManagerRemoved(blockManagerRemoved)`が呼び出される。

## 通知テンプレート

### メール通知の場合

該当なし。本通知はSpark内部のイベントバスを通じたプログラム間通知であり、メールやSMSなどの外部通知チャネルは使用しない。

### 本文テンプレート

```
イベント種別: SparkListenerBlockManagerRemoved
JSON形式:
{
  "Event": "SparkListenerBlockManagerRemoved",
  "Block Manager ID": {
    "Executor ID": "<executorId>",
    "Host": "<host>",
    "Port": <port>
  },
  "Timestamp": <timestamp>
}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | BlockManager削除時刻（ミリ秒） | System.currentTimeMillis() | Yes |
| blockManagerId | 削除されたBlockManagerの識別子 | BlockManagerMasterEndpoint内部マップ | Yes |
| blockManagerId.executorId | Executor ID | BlockManagerId.executorId | Yes |
| blockManagerId.host | ホスト名 | BlockManagerId.host | Yes |
| blockManagerId.port | ポート番号 | BlockManagerId.port | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | BlockManager削除 | BlockManagerMasterEndpointでremoveBlockManagerが呼ばれた場合 | Executorの離脱・タイムアウト・明示的削除時 |
| 内部処理 | Executorハートビートタイムアウト | ハートビートが一定時間途絶えた場合 | BlockManagerの自動クリーンアップ |
| 内部処理 | Executorの明示的削除 | scheduler.executorLost呼び出し後 | 動的リソース割り当てやExecutorのkill時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| LiveListenerBus停止済み | LiveListenerBusのstop()が呼ばれた後はイベントがドロップされる |
| BlockManager未登録 | 対象のBlockManagerがBlockManagerMasterEndpointに登録されていない場合は削除処理自体が行われない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[BlockManager削除要求発生] --> B[BlockManagerMasterEndpoint.removeBlockManager]
    B --> C[blockManagerInfo/blockManagerIdByExecutorからエントリ削除]
    C --> D[関連ブロック情報のクリーンアップ]
    D --> E[listenerBus.post SparkListenerBlockManagerRemoved]
    E --> F[LiveListenerBus.post]
    F --> G[各AsyncEventQueueにイベント配信]
    G --> H[SparkListenerBus.doPostEvent]
    H --> I[listener.onBlockManagerRemoved呼び出し]
    I --> J{受信リスナー}
    J -->|AppStatusListener| K[UI情報更新]
    J -->|EventLoggingListener| L[イベントログ書き込み]
    J -->|カスタムリスナー| M[ユーザ定義処理]
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| blockManagerInfo（インメモリMap） | 削除対象BlockManagerの情報参照 | BlockManagerMasterEndpoint内部のHashMap |
| blockManagerIdByExecutor（インメモリMap） | ExecutorIDからBlockManagerIDへのマッピング | BlockManagerMasterEndpoint内部のmutable.HashMap |

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

#### blockManagerInfo

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| BlockManagerId | 削除対象の特定 | キーとして指定されたblockManagerIdで検索 |
| BlockManagerInfo | ブロック情報のクリーンアップ | blockManagerInfoマップから取得 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| blockManagerInfo（インメモリMap） | DELETE | 対象BlockManagerのエントリ削除 |
| blockManagerIdByExecutor（インメモリMap） | DELETE | ExecutorIDマッピングの削除 |
| KVStore（AppStatusListener経由） | UPDATE | Executor情報のステータス更新 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| INSERT | Event | SparkListenerBlockManagerRemoved | EventLoggingListenerによりイベントログファイルに書き込み |
| INSERT | Timestamp | System.currentTimeMillis() | イベント発生時刻 |
| INSERT | Block Manager ID | blockManagerId | 削除されたBlockManagerの識別子 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| キュー満杯 | AsyncEventQueueのキャパシティ超過 | イベントがドロップされ、ドロップカウンターが増加。ログに警告出力 |
| リスナー例外 | リスナーのonBlockManagerRemovedで例外発生 | ListenerBusが例外をキャッチしログに記録。他のリスナーへの配信は継続 |
| バス停止後投稿 | LiveListenerBus.stop()後にpostが呼ばれた場合 | イベントは無視される（静かにドロップ） |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（イベントバスのキューサイズによる暗黙的な制限あり、デフォルト10000） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。Sparkアプリケーションの実行中、BlockManagerの削除が発生した任意のタイミングで配信される。

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

本イベントに含まれるBlockManagerId（ホスト名、ポート番号、ExecutorID）はクラスタ内部の情報であり、Spark UIやイベントログを通じて公開される。Spark UIのアクセス制御（ACL）やイベントログファイルのファイルシステムパーミッションにより、適切なアクセス制限が必要である。個人情報は含まれない。

## 備考

- SparkListenerBlockManagerRemovedはSparkListenerBlockManagerAddedと対になるイベントであり、BlockManagerのライフサイクルを追跡するために使用される。
- History Serverでのイベントログ再生時にも同様にリスナーに配信される。
- JsonProtocol.scalaにてJSON形式へのシリアライズ・デシリアライズが実装されている。

---

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

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

### 推奨読解順序

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

まず、SparkListenerBlockManagerRemovedイベントのデータ構造とBlockManagerIdの構成を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 112行目: case class SparkListenerBlockManagerRemovedの定義。time（Long）とblockManagerId（BlockManagerId）の2フィールドを持つ |
| 1-2 | BlockManagerId.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerId.scala` | BlockManagerIdのデータ構造（executorId, host, port）を理解する |

**読解のコツ**: SparkListenerEventを継承するcase classとして定義されており、パターンマッチングでの型判定とフィールドアクセスが行われる。

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

イベントの発火元であるBlockManagerMasterEndpointを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BlockManagerMasterEndpoint.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala` | 544行目: removeBlockManagerメソッド内でlistenerBus.postが呼び出される。ブロック情報のクリーンアップ後にイベントが発火される |

**主要処理フロー**:
1. **544行目**: `listenerBus.post(SparkListenerBlockManagerRemoved(System.currentTimeMillis(), blockManagerId))` -- イベントの生成と投稿

#### Step 3: イベントバスの配信メカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | 126行目: postメソッド。イベントを全てのAsyncEventQueueに配信する |
| 3-2 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 50行目: doPostEventでSparkListenerBlockManagerRemovedをパターンマッチしてlistener.onBlockManagerRemovedを呼び出す |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | 899行目: onBlockManagerRemovedでUI表示用のデータを更新する |
| 4-2 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 170行目: onBlockManagerRemovedでイベントをJSONとしてログに記録する |
| 4-3 | JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | 115行目, 285行目, 1114行目: イベントのJSON変換処理 |

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

```
BlockManagerMasterEndpoint.removeBlockManager()
    |
    +-- blockManagerInfo.remove(blockManagerId)
    +-- blockManagerIdByExecutor.remove(executorId)
    +-- listenerBus.post(SparkListenerBlockManagerRemoved)
            |
            +-- LiveListenerBus.post(event)
                    |
                    +-- postToQueues(event)
                            |
                            +-- AsyncEventQueue.post(event)
                                    |
                                    +-- SparkListenerBus.doPostEvent(listener, event)
                                            |
                                            +-- listener.onBlockManagerRemoved(event)
                                                    |
                                                    +-- AppStatusListener.onBlockManagerRemoved()
                                                    +-- EventLoggingListener.onBlockManagerRemoved()
```

### データフロー図

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

BlockManager削除要求 -----> BlockManagerMasterEndpoint -----> SparkListenerBlockManagerRemoved
(executorId,                  .removeBlockManager()              イベント
 blockManagerId)                    |
                                    v
                              LiveListenerBus.post() ---------> AsyncEventQueue
                                                                     |
                                                                     v
                                                              AppStatusListener --> KVStore (UI更新)
                                                              EventLoggingListener --> イベントログファイル
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントcase classの定義とリスナーインターフェース |
| BlockManagerMasterEndpoint.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala` | ソース | イベントの発火元（removeBlockManager） |
| 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用データの更新 |
| 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形式のシリアライズ・デシリアライズ |
| BasicEventFilterBuilder.scala | `core/src/main/scala/org/apache/spark/deploy/history/BasicEventFilterBuilder.scala` | ソース | History Serverでのイベントフィルタリング |
| CoarseGrainedSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala` | ソース | Executorの削除時にBlockManagerの削除が連鎖する |
