# 通知設計書 101-SparkListenerThriftServerOperationTimeout

## 概要

本ドキュメントは、Apache Spark HiveThriftServer2における `SparkListenerThriftServerOperationTimeout` イベント通知の設計を記述する。このイベントは、ThriftServerで実行中のオペレーション（SQLクエリ）がタイムアウトした際に発火し、LiveListenerBus経由で非同期配信される。

### 本通知の処理概要

ThriftServerオペレーションのタイムアウトを検知し、対応するリスナーへ通知するイベント機構である。SparkExecuteStatementOperationがクエリのタイムアウトを検出した時点でHiveThriftServer2EventManagerを通じてイベントを生成し、LiveListenerBusへポストする。HiveThriftServer2Listenerがイベントを受信し、オペレーションの実行状態をTIMEDOUTに遷移させ、KVStoreに永続化する。この仕組みにより、ThriftServer UIでのタイムアウトオペレーション表示やステータスモニタリングが実現される。

**業務上の目的・背景**：HiveThriftServer2はJDBC/ODBCクライアントからのSQLリクエストを処理するサーバーであり、長時間実行されるクエリがリソースを占有し続けることを防止するためにタイムアウト機構が必要である。本通知は、クエリがタイムアウトした事実をシステム全体に伝搬させ、UI表示やログ記録、リソース解放の契機とすることを目的とする。

**通知の送信タイミング**：SparkExecuteStatementOperationにおいて、設定されたタイムアウト時間（`spark.sql.thriftServer.queryTimeout` または個別クエリのqueryTimeout）が経過し、オペレーションが非終端状態にある場合に、`timeoutCancel()`メソッド内でイベントが発火する。具体的には、ScheduledExecutorServiceによるタイムアウトスレッドがオペレーションの状態をTIMEDOUTに遷移させた直後に送信される。

**通知の受信者**：SparkのLiveListenerBusに登録されたすべてのSparkListenerが受信対象となる。主たる受信者はHiveThriftServer2Listenerであり、KVStoreへの状態永続化を行う。さらに、カスタムリスナーを登録すればアプリケーション固有の処理も可能である。

**通知内容の概要**：オペレーションID（id）とタイムアウト発生時刻（finishTime）の2フィールドを含む。オペレーションIDによりどのクエリがタイムアウトしたかを特定でき、finishTimeにより発生時刻を記録できる。

**期待されるアクション**：受信者はオペレーションの状態をTIMEDOUTに更新し、UI上でタイムアウトしたオペレーションとして表示する。また、関連するSparkジョブのキャンセルやリソース解放が期待される。管理者はUI上でタイムアウトの頻度を監視し、タイムアウト設定値の調整やクエリの最適化を検討する。

## 通知種別

Sparkアプリケーション内部イベント通知（LiveListenerBus経由の非同期イベント配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中（通常のイベント配信優先度） |
| リトライ | なし（LiveListenerBusのイベントキューによる配信保証） |

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

LiveListenerBusのstatusキューに登録されたすべてのSparkListenerに対してブロードキャスト配信される。HiveThriftServer2Listenerは`onOtherEvent`メソッドでイベント型をパターンマッチし、`SparkListenerThriftServerOperationTimeout`の場合に`onOperationTimeout`を呼び出す。

## 通知テンプレート

### メール通知の場合

本通知はメール形式ではなく、Sparkの内部イベントバス経由のプログラム内通知であるため、メールテンプレートは該当しない。

### 本文テンプレート

```
SparkListenerThriftServerOperationTimeout(id={オペレーションID}, finishTime={タイムアウト発生時刻（ミリ秒エポック）})
```

### 添付ファイル

| ファイル名 | 形式 | 条件 | 説明 |
|----------|------|------|------|
| なし | - | - | 内部イベントのため添付ファイルは存在しない |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| id | オペレーション識別子（UUID文字列） | SparkExecuteStatementOperation.statementId | Yes |
| finishTime | タイムアウト発生時刻（ミリ秒エポック） | System.currentTimeMillis() | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| スケジュール（タイムアウト） | ScheduledExecutorService.schedule | オペレーションが非終端状態であること | SparkExecuteStatementOperation.timeoutCancel()内で、タイムアウト時間経過後にスケジュールされたRunnableが実行される |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| オペレーションが既に終端状態（FINISHED, CANCELED, ERROR, CLOSED等）にある場合 | timeoutCancel()内のsynchronizedブロックで`getStatus.getState.isTerminal`がtrueの場合は通知しない |
| タイムアウト値が0以下の場合 | timeout <= 0 の場合、timeoutExecutorそのものが生成されないため通知は発生しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[クエリ実行開始 - runInternal] --> B{timeout > 0?}
    B -->|Yes| C[ScheduledExecutorService生成]
    B -->|No| D[タイムアウト監視なし]
    C --> E[timeout秒後にRunnableを実行]
    E --> F[timeoutCancel呼び出し]
    F --> G{オペレーションが非終端状態?}
    G -->|Yes| H[状態をTIMEDOUTに変更]
    H --> I[cleanup実行]
    I --> J[eventManager.onStatementTimeout]
    J --> K[SparkListenerThriftServerOperationTimeout生成]
    K --> L[LiveListenerBus.post]
    L --> M[HiveThriftServer2Listener.onOtherEvent]
    M --> N[onOperationTimeout]
    N --> O[executionData.state = TIMEDOUT]
    O --> P[KVStore更新]
    G -->|No| Q[処理スキップ]
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| executionList（ConcurrentHashMap） | オペレーション実行データの参照 | インメモリ管理。オペレーションIDをキーとする |

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

#### executionList（ConcurrentHashMap）

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| LiveExecutionData | オペレーション実行状態の参照と更新 | executionList.get(e.id) で取得 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| KVStore（ElementTrackingStore） | UPDATE | ExecutionInfoのfinishTimestampとstateを更新 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | finishTimestamp | e.finishTime | タイムアウト発生時刻 |
| UPDATE | state | ExecutionState.TIMEDOUT | タイムアウト状態 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 不明なオペレーションID | executionListにオペレーションIDが存在しない場合 | logWarningでログ出力し、処理をスキップ |
| タイムアウトキャンセル中の例外 | timeoutCancel()内でNonFatalな例外が発生した場合 | HiveSQLExceptionとしてoperationExceptionに設定し、logErrorで記録 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（イベント発生ごとに配信） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

配信時間帯の制限なし。オペレーションのタイムアウトが発生した時点で即時配信される。

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

- 本イベントはprivate[thriftserver]スコープであり、thriftserverパッケージ外からのアクセスは制限されている
- オペレーションIDのみを含み、SQLステートメント本文やユーザー情報は直接含まない
- LiveListenerBusを経由するため、リスナー登録時のアクセス制御に依存する

## 備考

- タイムアウト値はグローバル設定（`spark.sql.thriftServer.queryTimeout`）と個別クエリ設定の小さい方が採用される
- SparkExecuteStatementOperation.timeoutCancel()のsynchronizedブロックにより、同一オペレーションに対する重複通知は防止される
- HiveThriftServer2Listener.onOperationTimeoutのログメッセージに「onOperationCanceled」と記載されている箇所があるが、これは既知の軽微な不整合である（実際にはタイムアウト処理が正しく実行される）

---

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

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

### 推奨読解順序

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

まず、イベントクラスとオペレーション実行データの構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 103-104行目: SparkListenerThriftServerOperationTimeoutケースクラスの定義（id, finishTimeフィールド） |
| 1-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 316-346行目: LiveExecutionDataクラスの定義。finishTimestamp, state等のフィールドを確認 |
| 1-3 | HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | 99-132行目: ExecutionInfoクラスの定義。isExecutionActiveでTIMEDOUTが非アクティブとして扱われる |

**読解のコツ**: case classの定義でextends SparkListenerEventを確認し、LiveListenerBusで配信可能なイベントであることを把握する。

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

タイムアウトの発生元であるSparkExecuteStatementOperationを起点として理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | 56-63行目: timeout値の決定ロジック。138-155行目: ScheduledExecutorServiceの生成とタイムアウトスケジュール |

**主要処理フロー**:
1. **56-63行目**: グローバルタイムアウトと個別タイムアウトの比較によるtimeout値決定
2. **138-155行目**: timeout > 0の場合、ScheduledExecutorServiceを生成してtimeout秒後にRunnableを実行
3. **284-295行目**: timeoutCancel()メソッド。synchronizedブロック内で非終端状態を確認後、状態をTIMEDOUTに変更し、cleanup()とeventManager.onStatementTimeout()を呼び出す

#### Step 3: イベント発行層を理解する

HiveThriftServer2EventManagerによるイベント生成とLiveListenerBusへのポストを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 60-62行目: onStatementTimeoutメソッド。SparkListenerThriftServerOperationTimeoutインスタンスを生成しpostLiveListenerBusへ委譲 |
| 3-2 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 28-30行目: postLiveListenerBusメソッド。sc.listenerBus.post(event)でLiveListenerBusにポスト |

#### Step 4: イベント受信層を理解する

HiveThriftServer2Listenerによるイベント受信と状態更新を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 117-130行目: onOtherEventメソッド。イベント型のパターンマッチで124行目がSparkListenerThriftServerOperationTimeoutに対応 |
| 4-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 194-203行目: onOperationTimeoutメソッド。executionListからデータ取得、finishTimestampとstateをTIMEDOUTに設定し、updateLiveStoreでKVStoreを更新 |

**主要処理フロー**:
- **194行目**: executionList.get(e.id)でOption取得
- **197行目**: finishTimestamp = e.finishTime
- **198行目**: state = ExecutionState.TIMEDOUT
- **199行目**: updateLiveStore(executionData)

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

```
SparkExecuteStatementOperation.timeoutCancel() [行284-295]
    |
    +-- OperationState.TIMEDOUT に setState [行290]
    |
    +-- cleanup() [行291, 行308-324]
    |       |
    |       +-- backgroundHandle.cancel(true) [行311]
    |       +-- sparkContext.cancelJobGroup(statementId) [行317]
    |
    +-- HiveThriftServer2.eventManager.onStatementTimeout(statementId) [行292]
            |
            +-- HiveThriftServer2EventManager.onStatementTimeout(id) [行60-62]
                    |
                    +-- postLiveListenerBus(SparkListenerThriftServerOperationTimeout(id, time)) [行61]
                            |
                            +-- sc.listenerBus.post(event) [行29]
                                    |
                                    +-- HiveThriftServer2Listener.onOtherEvent(event) [行117]
                                            |
                                            +-- onOperationTimeout(e) [行124, 行194-203]
                                                    |
                                                    +-- executionData.state = ExecutionState.TIMEDOUT [行198]
                                                    +-- updateLiveStore(executionData) [行199]
```

### データフロー図

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

ScheduledExecutorService  --->  timeoutCancel()               --->  SparkListenerThriftServerOperationTimeout
(timeout秒経過)                  |                                    |
                                +-- statementId              --->  LiveListenerBus.post()
                                +-- System.currentTimeMillis()       |
                                                                     +---> HiveThriftServer2Listener
                                                                            |
                                                                            +---> ExecutionInfo (KVStore)
                                                                            |     state=TIMEDOUT
                                                                            |     finishTimestamp=finishTime
                                                                            |
                                                                            +---> ThriftServer UI (表示)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | ソース | イベント定義・生成・LiveListenerBusへのポスト |
| HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | ソース | イベント受信・状態更新・KVStore永続化 |
| SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | ソース | タイムアウト発生元。timeout値の決定とtimeoutCancel()の実装 |
| HiveThriftServer2.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala` | ソース | eventManagerの初期化。ExecutionState列挙体の定義（TIMEDOUT含む） |
| HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | ソース | KVStore経由のステータス参照。ExecutionInfoの定義 |
| SparkOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala` | ソース | SparkOperationトレイト。close()でonOperationClosedを呼び出す |
| HiveThriftServer2ListenerSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2ListenerSuite.scala` | テスト | タイムアウトイベントの処理を含むリスナーテスト（165行目） |
