# 通知設計書 35-SparkListenerNodeBlacklisted

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerNodeBlacklistedイベント通知の設計について記述する。本イベントはノード（ホスト）がアプリケーションレベルでブラックリスト（除外）された際に発火する非推奨イベントであり、Spark 3.1.0以降はSparkListenerNodeExcludedに置き換えられている。

### 本通知の処理概要

SparkListenerNodeBlacklistedは、あるノード上で複数のExecutorが除外された結果、ノード全体がアプリケーションレベルで除外された際に発火するイベント通知である。

**業務上の目的・背景**：単一ノード上の複数Executorが繰り返し問題を起こしている場合、個別のExecutor除外では不十分であり、ノード全体を除外する必要がある。これはハードウェア故障、ディスク障害、ネットワーク障害などノードレベルの問題を示すことが多い。本通知はノード除外が発生したことをリスナーに通知する。

**通知の送信タイミング**：(1) HealthTracker.updateExcludedForSuccessfulTaskSet内で、ノード上の除外Executor数が`spark.excludeOnFailure.application.maxFailedExecutorsPerNode`の閾値に達した場合、(2) HealthTracker.updateExcludedForFetchFailure内で、外部シャッフルサービス有効時にフェッチ失敗が発生した場合。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装。

**通知内容の概要**：除外されたノードのホストID、除外発生時刻、除外Executor数（またはフェッチ失敗数）が含まれる。

**期待されるアクション**：Spark UIでノード除外状態が表示される。運用者はノードレベルの問題を認識し、ハードウェア点検やノードの入替を検討する。

## 通知種別

アプリ内通知（Sparkイベントバスによるインプロセス非同期イベント配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 高（ノード除外はExecutor除外より重大） |
| リトライ | なし |

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

LiveListenerBusに登録された全てのSparkListenerInterface実装に対してブロードキャスト配信される。SparkListenerBus.doPostEventのパターンマッチにより`onNodeBlacklisted`メソッドが呼び出される。

## 通知テンプレート

### イベントオブジェクト

| 項目 | 内容 |
|-----|------|
| イベントクラス | `SparkListenerNodeBlacklisted` |
| パッケージ | `org.apache.spark.scheduler` |
| 親クラス | `SparkListenerEvent` |
| シリアライズ形式 | JSON（JsonProtocol経由） |

### イベントデータ構造

```scala
@deprecated("use SparkListenerNodeExcluded instead", "3.1.0")
case class SparkListenerNodeBlacklisted(
    time: Long,
    hostId: String,
    executorFailures: Int)
  extends SparkListenerEvent
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | イベント発生時刻（ミリ秒） | `clock.getTimeMillis()` | Yes |
| hostId | 除外対象のノードホスト名 | HealthTracker内部管理 | Yes |
| executorFailures | 除外Executor数 | excludedExecsOnNode.size または 1（フェッチ失敗時） | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | TaskSet正常完了後のExecutor除外 | ノード上の除外Executor数 >= MAX_FAILED_EXEC_PER_NODE | HealthTracker.updateExcludedForSuccessfulTaskSet内 |
| 内部処理 | フェッチ失敗発生 | EXCLUDE_FETCH_FAILURE_ENABLED=true かつ SHUFFLE_SERVICE_ENABLED=true | HealthTracker.updateExcludedForFetchFailure内 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 既にノード除外済み | nodeIdToExcludedExpiryTimeに既に登録されているノードは再除外されない |
| excludeOnFailure無効 | HealthTrackerが生成されない場合はイベント自体が発生しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[Executor除外発生] --> B[ノード上の除外Executor数を確認]
    B --> C{除外Executor数 >= MAX_FAILED_EXEC_PER_NODE?}
    C -->|Yes| D{既にノード除外済み?}
    C -->|No| Z[処理終了]
    D -->|No| E[nodeIdToExcludedExpiryTimeに登録]
    D -->|Yes| Z
    E --> F[SparkListenerNodeBlacklisted発行]
    E --> G[SparkListenerNodeExcluded発行]
    F --> H[LiveListenerBusでリスナーに配信]
    G --> H
    H --> I[_excludedNodeList更新]
    I --> J[killExecutorsOnExcludedNode実行]
    J --> Z
```

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

該当なし（インメモリのデータ構造を使用）

### インメモリデータ構造

| データ構造 | 用途 | 備考 |
|-----------|------|------|
| nodeIdToExcludedExpiryTime | ノード除外と有効期限の管理 | HashMap[String, Long] |
| nodeToExcludedExecs | ノード毎の除外Executor一覧 | HashMap[String, HashSet[String]] |
| _excludedNodeList | スレッドセーフな除外ノード集合 | AtomicReference[Set[String]] |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| イベント配信失敗 | ListenerBusキュー溢れ | LiveListenerBusのドロップポリシーに従う |
| Executor Kill失敗 | allocationClientが未定義 | ログ出力のみ（除外自体は維持） |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0 |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

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

### 配信時間帯

制限なし

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

- 本イベントはクラスタ内部の情報のみを含み、個人情報は含まない
- ノード除外時にExecutorのkillまたはdecommissionが実行される場合がある

## 備考

- 本イベントは`@deprecated`であり、SparkListenerNodeExcludedが推奨される
- ノード除外時にEXCLUDE_ON_FAILURE_KILL_ENABLEDが有効であれば、ノード上の全Executorがkillまたはdecommissionされる
- 外部シャッフルサービス有効時のフェッチ失敗では、executorFailuresは1固定でノードが即座に除外される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 194-200行目: SparkListenerNodeBlacklistedのcase class定義。time, hostId, executorFailuresを持つ |

**読解のコツ**: Executor除外イベントの`executorId`に対応するのがノード除外イベントの`hostId`。`executorFailures`はExecutor失敗「数」であり、taskFailuresとは異なる。

#### Step 2: エントリーポイントを理解する（TaskSet完了時）

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HealthTracker.scala | `core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala` | 264-315行目: updateExcludedForSuccessfulTaskSetメソッド。Executor除外後のノード除外判定 |

**主要処理フロー**:
1. **302行目**: `excludedExecsOnNode.size >= MAX_FAILED_EXEC_PER_NODE`でノード除外閾値判定
2. **309行目**: `listenerBus.post(SparkListenerNodeBlacklisted(now, node, excludedExecsOnNode.size))`
3. **310行目**: `listenerBus.post(SparkListenerNodeExcluded(now, node, excludedExecsOnNode.size))`
4. **311行目**: `_excludedNodeList.set(nodeIdToExcludedExpiryTime.keySet.toSet)`で除外ノード集合更新
5. **312行目**: `killExecutorsOnExcludedNode(node)`でノード上Executorのkill

#### Step 3: エントリーポイントを理解する（フェッチ失敗時）

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HealthTracker.scala | `core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala` | 220-244行目: updateExcludedForFetchFailure内のシャッフルサービス有効時処理 |

**主要処理フロー**:
- **240行目**: `listenerBus.post(SparkListenerNodeBlacklisted(now, host, 1))`

#### Step 4: イベント配信メカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 74-75行目: doPostEvent内でonNodeBlacklistedにルーティング |

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

```
TaskSchedulerImpl
    |
    +-- HealthTracker.updateExcludedForSuccessfulTaskSet()
    |       |
    |       +-- (Executor除外判定・実行)
    |       +-- excludedExecsOnNode.size >= MAX_FAILED_EXEC_PER_NODE
    |       +-- listenerBus.post(SparkListenerNodeBlacklisted)
    |       +-- listenerBus.post(SparkListenerNodeExcluded)
    |       +-- killExecutorsOnExcludedNode()
    |
    +-- HealthTracker.updateExcludedForFetchFailure()
            |
            +-- (シャッフルサービス有効時)
            +-- listenerBus.post(SparkListenerNodeBlacklisted)
            +-- listenerBus.post(SparkListenerNodeExcluded)
            +-- killExecutorsOnExcludedNode()
```

### データフロー図

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

Executor除外情報          HealthTracker                    SparkListenerNodeBlacklisted
(node, excludedExecs ---> .updateExcludedFor...()   --->   イベント
 OnNode.size)             ノード除外閾値判定                   |
                                                            v
                                                     LiveListenerBus
                                                            |
                                                     +------+------+
                                                     |             |
                                               AppStatus     EventLogging
                                               Listener      Listener
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントクラス定義 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ |
| HealthTracker.scala | `core/src/main/scala/org/apache/spark/scheduler/HealthTracker.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` | ソース | イベントログへの永続化 |
