# 通知設計書 102-SparkListenerThriftServerOperationError

## 概要

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

### 本通知の処理概要

ThriftServerオペレーションの実行エラーを検知し、エラー情報をシステム全体に伝搬するイベント機構である。SparkExecuteStatementOperationやSparkOperationトレイトのonError()で例外が捕捉された際に、HiveThriftServer2EventManager.onStatementError()を通じてイベントを生成する。イベントにはオペレーションID、エラーメッセージ、エラースタックトレース、エラー発生時刻が含まれ、HiveThriftServer2Listenerが受信してKVStoreに永続化する。

**業務上の目的・背景**：HiveThriftServer2はクライアントからのSQLリクエストを処理するサーバーであり、クエリ実行時のエラーを適切に記録・表示することは運用監視の基本要件である。本通知は、エラーが発生した事実と詳細情報（エラーメッセージ、スタックトレース）をシステム全体に伝搬させ、ThriftServer UIでのエラーオペレーション表示、ログ記録、管理者による障害分析を可能にすることを目的とする。

**通知の送信タイミング**：以下のいずれかの状況でイベントが発火する。(1) SparkExecuteStatementOperation.execute()内でクエリ実行中にThrowableが捕捉され、オペレーション状態がERRORに遷移した時。(2) SparkExecuteStatementOperation.runInternal()でバックグラウンド実行サブミット時にRejectedExecutionExceptionまたはNonFatal例外が発生した時。(3) SparkOperationトレイトのonError()メソッドでエラーが捕捉された時（各種オペレーション共通のエラーハンドリング）。

**通知の受信者**：SparkのLiveListenerBusに登録されたすべてのSparkListenerが受信対象となる。主たる受信者はHiveThriftServer2Listenerであり、KVStoreへのエラー情報の永続化を行う。

**通知内容の概要**：オペレーションID（id）、エラーメッセージ（errorMsg）、エラースタックトレース（errorTrace）、エラー発生時刻（finishTime）の4フィールドを含む。

**期待されるアクション**：受信者はオペレーションの状態をFAILEDに更新し、エラーメッセージをdetailフィールドに格納する。ThriftServer UI上でエラーオペレーションとして表示され、管理者はエラーの原因を分析し、SQLクエリやシステム設定の修正を行う。

## 通知種別

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

## 送信仕様

### 基本情報

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

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

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

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```
SparkListenerThriftServerOperationError(id={オペレーションID}, errorMsg={エラーメッセージ}, errorTrace={スタックトレース}, finishTime={エラー発生時刻（ミリ秒エポック）})
```

### 添付ファイル

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

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| id | オペレーション識別子（UUID文字列） | SparkExecuteStatementOperation.statementId | Yes |
| errorMsg | エラーメッセージ | e.getMessage() | Yes |
| errorTrace | エラースタックトレース文字列 | Utils.exceptionString(e) | Yes |
| finishTime | エラー発生時刻（ミリ秒エポック） | System.currentTimeMillis() | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 例外捕捉（execute内） | SparkExecuteStatementOperation.execute()でのThrowable | オペレーションが非終端状態であること | クエリ実行中の例外発生時（行246-272） |
| 例外捕捉（submit時） | SparkExecuteStatementOperation.runInternal()でのRejectedExecutionException | 常時 | バックグラウンドタスクのサブミット拒否時（行196-208） |
| 例外捕捉（共通エラー） | SparkOperation.onError() | 常時 | 各種オペレーション共通のエラーハンドリング（行115-126） |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| オペレーションが既に終端状態にある場合（execute内） | getStatus.getState.isTerminalがtrueの場合、logWarningのみ出力し通知は発生しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[クエリ実行中に例外発生] --> B{オペレーションが終端状態?}
    B -->|No| C[状態をERRORに変更]
    C --> D[eventManager.onStatementError呼び出し]
    D --> E[SparkListenerThriftServerOperationError生成]
    E --> F[LiveListenerBus.post]
    F --> G[HiveThriftServer2Listener.onOtherEvent]
    G --> H[onOperationError]
    H --> I[executionData.detail = errorMsg]
    I --> J[executionData.state = FAILED]
    J --> K[KVStore更新]
    B -->|Yes| L[logWarning出力のみ]
```

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

### 参照テーブル一覧

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

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

#### executionList（ConcurrentHashMap）

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

### 更新テーブル一覧

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

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | finishTimestamp | e.finishTime | エラー発生時刻 |
| UPDATE | detail | e.errorMsg | エラーメッセージ |
| UPDATE | state | ExecutionState.FAILED | 失敗状態 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 不明なオペレーションID | executionListにオペレーションIDが存在しない場合 | logWarningでログ出力し、処理をスキップ |
| HiveSQLException | クエリ実行中にHiveSQLExceptionが発生した場合 | onStatementError通知後、例外を再スローする |
| RejectedExecutionException | バックグラウンドタスクのサブミットが拒否された場合 | onStatementError通知後、taskExecutionRejectedErrorをスロー |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

配信時間帯の制限なし。エラーが発生した時点で即時配信される。

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

- 本イベントはprivate[thriftserver]スコープであり、thriftserverパッケージ外からのアクセスは制限されている
- errorMsgとerrorTraceにはスタックトレース情報が含まれるため、内部実装の詳細が漏洩する可能性がある。ただし、LiveListenerBus内部でのみ配信されるため外部への直接的な漏洩リスクは限定的である
- ThriftServer UIにエラー詳細が表示されるため、UIアクセス制御が重要

## 備考

- errorMsgはe.getMessage()から取得されるため、nullの可能性がある
- errorTraceはUtils.exceptionString(e)で生成され、完全なスタックトレースを含む
- SparkOperationトレイトのonError()は、SparkGetCatalogsOperation、SparkGetColumnsOperation、SparkGetFunctionsOperation等、複数のオペレーションクラスで共有されている

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 106-110行目: SparkListenerThriftServerOperationErrorケースクラスの定義（id, errorMsg, errorTrace, finishTimeフィールド） |
| 1-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 316-346行目: LiveExecutionDataクラス。detailフィールドにエラーメッセージが格納される |
| 1-3 | HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | 99-132行目: ExecutionInfoクラス。stateフィールドとisExecutionActiveの定義 |

**読解のコツ**: errorMsgとerrorTraceの2つのエラー情報フィールドがあることに注意。errorMsgはExecutionInfoのdetailに格納されるが、errorTraceはイベントフィールドとしてのみ存在しKVStoreには永続化されない。

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

エラー発生元のSparkExecuteStatementOperationおよびSparkOperationを起点として理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | 246-272行目: execute()内のcatchブロック。非終端状態の場合にonStatementErrorを呼び出す |
| 2-2 | SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | 196-208行目: runInternal()内のcatchブロック。サブミット拒否時・NonFatal例外時にonStatementErrorを呼び出す |
| 2-3 | SparkOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala` | 115-126行目: onError()メソッド。各種オペレーション共通のエラーハンドリング |

**主要処理フロー**:
1. **246行目**: execute()内でThrowableをキャッチ
2. **256行目**: getStatus().getState().isTerminal で終端状態チェック
3. **264行目**: setState(OperationState.ERROR) で状態をERRORに変更
4. **265-266行目**: eventManager.onStatementError(statementId, e.getMessage, Utils.exceptionString(e)) 呼び出し

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

HiveThriftServer2EventManagerによるイベント生成を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 64-67行目: onStatementErrorメソッド。id, errorMsg, errorTrace, System.currentTimeMillis()でイベント生成 |

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 125行目: onOtherEvent内のSparkListenerThriftServerOperationErrorパターンマッチ |
| 4-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 205-215行目: onOperationErrorメソッド。finishTimestamp, detail, stateをFAILEDに設定 |

**主要処理フロー**:
- **207行目**: executionData.finishTimestamp = e.finishTime
- **209行目**: executionData.detail = e.errorMsg
- **210行目**: executionData.state = ExecutionState.FAILED
- **211行目**: updateLiveStore(executionData)

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

```
SparkExecuteStatementOperation.execute() [行212-282]
    |
    +-- catch Throwable [行246]
            |
            +-- setState(OperationState.ERROR) [行264]
            |
            +-- HiveThriftServer2.eventManager.onStatementError(statementId, msg, trace) [行265-266]
                    |
                    +-- HiveThriftServer2EventManager.onStatementError(id, errorMsg, errorTrace) [行64-67]
                            |
                            +-- postLiveListenerBus(SparkListenerThriftServerOperationError(...)) [行65-66]
                                    |
                                    +-- sc.listenerBus.post(event) [行29]
                                            |
                                            +-- HiveThriftServer2Listener.onOtherEvent(event) [行117]
                                                    |
                                                    +-- onOperationError(e) [行125, 行205-215]
                                                            |
                                                            +-- executionData.state = ExecutionState.FAILED [行210]
                                                            +-- updateLiveStore(executionData) [行211]

SparkOperation.onError() [行115-126]  (別エントリーポイント)
    |
    +-- setState(OperationState.ERROR) [行119]
    |
    +-- HiveThriftServer2.eventManager.onStatementError(...) [行120-121]
```

### データフロー図

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

Throwable例外             --->  execute() catch/onError()          --->  SparkListenerThriftServerOperationError
  |                              |                                         |
  +-- e.getMessage()      --->  onStatementError(id, msg, trace)  --->  LiveListenerBus.post()
  +-- Utils.exceptionString(e)                                             |
  +-- System.currentTimeMillis()                                           +---> HiveThriftServer2Listener
                                                                                  |
                                                                                  +---> ExecutionInfo (KVStore)
                                                                                  |     state=FAILED
                                                                                  |     detail=errorMsg
                                                                                  |     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` | ソース | 主要なエラー発生元。execute()とrunInternal()でのエラーハンドリング |
| SparkOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala` | ソース | 共通エラーハンドリングのonError()メソッド |
| HiveThriftServer2.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala` | ソース | eventManagerの初期化。ExecutionState列挙体の定義 |
| HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | ソース | KVStore経由のステータス参照。ExecutionInfoの定義 |
| SparkGetCatalogsOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala` | ソース | onError()を使用するオペレーションの一例 |
| HiveThriftServer2ListenerSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2ListenerSuite.scala` | テスト | エラーイベントの処理を含むリスナーテスト（166-167行目） |
