# 通知設計書 21-SparkListenerUnschedulableTaskSetAdded

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerUnschedulableTaskSetAddedイベント通知の設計仕様を定義する。このイベントは、excludeOnFailure機能（旧blacklist機能）によりタスクセットがスケジュール不可能になった場合に発火し、動的リソース割り当てマネージャに対してExecutorの追加要求を促すために使用される。

### 本通知の処理概要

本通知は、Sparkのタスクスケジューリングにおいてタスクセットがスケジュール不可能な状態に遷移した際に、関連リスナーへ非同期で配信されるイベントである。

**業務上の目的・背景**：Sparkのタスク実行において、excludeOnFailure機能により障害のあるExecutorやノードが除外されると、一部のタスクセットがスケジュール可能なExecutorを失い、実行できなくなる場合がある。この通知は、そのような状態を動的リソース割り当て（Dynamic Allocation）マネージャに伝え、新しいExecutorの追加をトリガーすることで、ジョブの停滞を防止する役割を担う。

**通知の送信タイミング**：TaskSetManagerがスケジュール不可能な状態であることを検出し、DAGSchedulerのイベントキューにUnschedulableTaskSetAddedメッセージが投入された際に発火する。具体的には、excludeOnFailure機能が有効かつ動的割り当てが有効な環境で、利用可能なExecutorが不足した場合に発生する。

**通知の受信者**：LiveListenerBusに登録されたすべてのSparkListenerInterface実装クラスが受信対象となる。主要な受信者はExecutorAllocationManager内のリスナーであり、unschedulableTaskSetsセットへのステージ追加とExecutor追加要求（onSchedulerBacklogged）をトリガーする。

**通知内容の概要**：stageId（ステージID）およびstageAttemptId（ステージ試行ID）の2つのフィールドを含み、どのステージのどの試行がスケジュール不可能になったかを特定できる。

**期待されるアクション**：ExecutorAllocationManagerが新しいExecutorの追加を要求し、スケジュール不可能なタスクセットが再びスケジュール可能になることを目指す。運用者はSpark UIまたはイベントログを通じてスケジュール不可能なタスクセットの発生状況を監視できる。

## 通知種別

アプリ内イベント通知（Sparkイベントリスナーバス経由の非同期配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中 |
| リトライ | 無（イベントバスのキュー管理に依存） |

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

LiveListenerBusに登録されたすべてのSparkListenerInterface実装リスナーに対してブロードキャスト配信される。SparkListenerBus.doPostEventメソッドにおいて、SparkListenerUnschedulableTaskSetAddedイベントとパターンマッチし、listener.onUnschedulableTaskSetAdded()が呼び出される。

## 通知テンプレート

### メール通知の場合

該当なし。本通知はSparkの内部イベントバスを介したプログラマティック通知であり、メール等の外部通知チャネルは使用しない。

### 本文テンプレート

```
イベント: SparkListenerUnschedulableTaskSetAdded
ステージID: {stageId}
ステージ試行ID: {stageAttemptId}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| stageId | スケジュール不可能なタスクセットが属するステージのID | DAGSchedulerイベント | Yes |
| stageAttemptId | ステージの試行回数ID | DAGSchedulerイベント | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| スケジューライベント | TaskSetManagerがスケジュール不可判定 | excludeOnFailureが有効かつ動的割り当て有効 | TaskSetManagerがスケジュール可能なExecutorを検出できない場合にDAGSchedulerへ通知 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 動的割り当て無効 | spark.dynamicAllocation.enabledがfalseの場合、本イベントは発火しない |
| excludeOnFailure無効 | spark.excludeOnFailure.enabledがfalseの場合、Executorの除外が発生しないため本イベントも発火しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[TaskSetManager: スケジュール不可を検出] --> B[DAGSchedulerEventProcessLoop: UnschedulableTaskSetAdded]
    B --> C[DAGScheduler.handleUnschedulableTaskSetAdded]
    C --> D[listenerBus.post SparkListenerUnschedulableTaskSetAdded]
    D --> E[SparkListenerBus.doPostEvent]
    E --> F[listener.onUnschedulableTaskSetAdded]
    F --> G[ExecutorAllocationManager: unschedulableTaskSets追加]
    G --> H[onSchedulerBacklogged: Executor追加要求]
```

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

### 参照テーブル一覧

該当なし。本通知はインメモリのイベントバスを使用し、データベースへのアクセスは行わない。

### 更新テーブル一覧

該当なし。ただし、EventLoggingListenerが有効な場合はイベントログファイル（JSON形式）に書き込まれる。

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonUnschedulableTaskSetAdded実装で例外発生 | ListenerBusが例外をキャッチしログ出力。他リスナーへの配信は継続 |
| イベントキュー溢れ | LiveListenerBusのキューが満杯 | イベントがドロップされる。spark.scheduler.listenerbus.eventqueue.capacityで調整可能 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（イベント発生頻度に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。Sparkアプリケーションの実行中、条件が満たされれば随時発火する。

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

本通知にはステージIDとステージ試行IDのみが含まれ、個人情報や機密データは含まれない。イベントログへの書き込み時にはファイルシステムのアクセス制御に依存する。

## 備考

- Since 3.1.0で導入されたイベントである。
- 従来のblacklist機能はexcludeOnFailureにリネームされた（SPARK-32037）。
- 本イベントはSparkListenerBusのdoPostEventにおいて専用のcaseブランチで処理される（SparkListenerBus.scala 行94-95）。
- SparkListenerLogStartやSparkListenerMiscellaneousProcessAddedとは異なり、本イベントはonOtherEventではなく専用のリスナーメソッドで処理される。

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 行223-225: SparkListenerUnschedulableTaskSetAddedケースクラスの定義。stageIdとstageAttemptIdの2フィールドを持つ |

**読解のコツ**: SparkListenerEventを継承するケースクラスとして定義されており、@Since("3.1.0")アノテーションが付与されている。

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

タスクセットがスケジュール不可能と判定される起点を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | 行3560-3561: DAGSchedulerEventProcessLoopにおけるUnschedulableTaskSetAddedメッセージの処理 |
| 2-2 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | 行1302-1305: handleUnschedulableTaskSetAddedメソッド。listenerBus.postでイベントを発火 |

**主要処理フロー**:
1. **行3560**: DAGSchedulerEventProcessLoopがUnschedulableTaskSetAddedメッセージを受信
2. **行3561**: dagScheduler.handleUnschedulableTaskSetAdded(stageId, stageAttemptId)を呼び出し
3. **行1305**: listenerBus.post(SparkListenerUnschedulableTaskSetAdded(stageId, stageAttemptId))でイベント発火

#### Step 3: イベント配信層を理解する

LiveListenerBusを経由してリスナーに配信される仕組みを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 行94-95: doPostEventでSparkListenerUnschedulableTaskSetAddedをパターンマッチし、listener.onUnschedulableTaskSetAddedを呼び出す |

#### Step 4: 主要リスナーの処理を理解する

ExecutorAllocationManagerでの処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ExecutorAllocationManager.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | 行842-850: onUnschedulableTaskSetAddedのオーバーライド。unschedulableTaskSetsにステージを追加し、onSchedulerBackloggedを呼び出す |

**主要処理フロー**:
- **行844**: stageIdを取得
- **行845**: stageAttemptIdを取得
- **行846**: StageAttemptオブジェクトを生成
- **行848**: unschedulableTaskSets.add(stageAttempt)でセットに追加
- **行849**: allocationManager.onSchedulerBacklogged()でExecutor追加要求

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

```
TaskSetManager (スケジュール不可検出)
    |
    +-- DAGSchedulerEventProcessLoop.onReceive
           |
           +-- UnschedulableTaskSetAdded(stageId, stageAttemptId)
                  |
                  +-- DAGScheduler.handleUnschedulableTaskSetAdded
                         |
                         +-- listenerBus.post(SparkListenerUnschedulableTaskSetAdded)
                                |
                                +-- SparkListenerBus.doPostEvent
                                       |
                                       +-- listener.onUnschedulableTaskSetAdded
                                              |
                                              +-- ExecutorAllocationManager.onUnschedulableTaskSetAdded
                                                     |
                                                     +-- onSchedulerBacklogged()
```

### データフロー図

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

stageId, stageAttemptId ------> DAGScheduler.handleUnschedulableTaskSetAdded ---> SparkListenerUnschedulableTaskSetAdded
                                       |
                                       v
                                LiveListenerBus (非同期キュー)
                                       |
                                       v
                                ExecutorAllocationManager ---------------------> Executor追加要求
                                EventLoggingListener --------------------------> イベントログ (JSON)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントケースクラスおよびリスナーインターフェース定義 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベント配信のディスパッチロジック |
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | イベント発火元（handleUnschedulableTaskSetAdded） |
| ExecutorAllocationManager.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | ソース | 主要リスナー。Executor動的追加のトリガー |
| 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直列化・逆直列化 |
