# 通知設計書 103-SparkListenerThriftServerOperationFinish

## 概要

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

### 本通知の処理概要

ThriftServerオペレーションの正常完了を検知し、対応するリスナーへ通知するイベント機構である。SparkExecuteStatementOperationのexecute()メソッド内で、クエリ実行が正常に完了し、オペレーション状態がFINISHEDに遷移した直後にHiveThriftServer2EventManager.onStatementFinish()を通じてイベントが生成される。HiveThriftServer2Listenerがイベントを受信し、オペレーションの実行状態をFINISHEDに遷移させ、KVStoreに永続化する。

**業務上の目的・背景**：HiveThriftServer2はクライアントからのSQLリクエストを処理するサーバーであり、クエリが正常に完了したことを記録・追跡することは、パフォーマンスモニタリングや実行履歴管理の基本要件である。本通知は、クエリ実行が正常に完了した事実をシステム全体に伝搬させ、ThriftServer UIでのオペレーション完了表示、実行時間の計測、ジョブ履歴の管理を可能にすることを目的とする。

**通知の送信タイミング**：SparkExecuteStatementOperation.execute()メソッドのfinallyブロック内で、オペレーションが非終端状態にある場合に送信される。具体的には、SQLクエリの実行（session.sql()）とデータ収集（result.collect()またはtoLocalIterator()）が完了した後、synchronizedブロック内でsetState(OperationState.FINISHED)を呼び出した直後にeventManager.onStatementFinish()が呼び出される。

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

**通知内容の概要**：オペレーションID（id）と完了時刻（finishTime）の2フィールドを含む。

**期待されるアクション**：受信者はオペレーションの状態をFINISHEDに更新する。ThriftServer UI上で完了したオペレーションとして表示され、実行時間の計算（startTimestampからfinishTimestampまでの差分）が可能になる。なお、FINISHED状態のオペレーションはまだアクティブとして扱われ（isExecutionActive=true）、後続のOperationClosed通知でクローズされるまでアクティブカウントに含まれる。

## 通知種別

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

## 送信仕様

### 基本情報

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

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

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

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```
SparkListenerThriftServerOperationFinish(id={オペレーションID}, finishTime={完了時刻（ミリ秒エポック）})
```

### 添付ファイル

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

## テンプレート変数

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

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| クエリ実行完了 | SparkExecuteStatementOperation.execute()のfinallyブロック | オペレーションが非終端状態であること | synchronizedブロック内でFINISHEDに遷移後にイベント発行（行274-279） |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| オペレーションが既に終端状態にある場合 | execute()のfinallyブロック内のsynchronizedで`getStatus.getState.isTerminal`がtrueの場合、setState(FINISHED)とonStatementFinish()のいずれも実行されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[execute開始] --> B[session.sql実行]
    B --> C[結果データ収集]
    C --> D[finallyブロック]
    D --> E{synchronized: オペレーションが非終端状態?}
    E -->|Yes| F[setState - FINISHED]
    F --> G[eventManager.onStatementFinish]
    G --> H[SparkListenerThriftServerOperationFinish生成]
    H --> I[LiveListenerBus.post]
    I --> J[HiveThriftServer2Listener.onOtherEvent]
    J --> K[onOperationFinished]
    K --> L[executionData.state = FINISHED]
    L --> M[KVStore更新]
    E -->|No| N[処理スキップ - 既にERROR/CANCELED/TIMEDOUT等]
```

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

### 参照テーブル一覧

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

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

#### executionList（ConcurrentHashMap）

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

### 更新テーブル一覧

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

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | finishTimestamp | e.finishTime | 完了時刻 |
| UPDATE | state | ExecutionState.FINISHED | 完了状態 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 不明なオペレーションID | executionListにオペレーションIDが存在しない場合 | logWarningでログ出力し、処理をスキップ |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

配信時間帯の制限なし。クエリが正常完了した時点で即時配信される。

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

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

## 備考

- FINISHED状態はオペレーションのライフサイクルにおける中間状態である。FINISHEDの後にOperationClosedイベントが発火し、最終的にCLOSED状態に遷移する
- isExecutionActiveの定義では、FINISHEDはアクティブとして扱われる（FAILED, CANCELED, TIMEDOUT, CLOSEDのみが非アクティブ）。これは、FINISHED後もクライアントが結果を取得中の場合があるためである
- execute()メソッドのfinallyブロック内でsynchronizedが使用されているため、エラー発生時のキャンセル処理やタイムアウト処理と排他的に動作する

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 112-114行目: SparkListenerThriftServerOperationFinishケースクラスの定義（id, finishTimeフィールド） |
| 1-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 316-346行目: LiveExecutionDataクラスの定義 |
| 1-3 | HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | 117-123行目: isExecutionActiveの定義。FINISHEDはアクティブとして扱われることを確認 |

**読解のコツ**: FINISHEDがisExecutionActiveでtrueを返す点は重要。FINISHED状態はクエリ実行完了を意味するが、クライアントによる結果取得はまだ進行中の可能性があるため、CLOSEDまでアクティブとみなされる。

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

クエリ完了時の通知発行元を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | 273-281行目: execute()のfinallyブロック。synchronizedブロック内で非終端状態確認後にFINISHEDに遷移しonStatementFinishを呼び出す |

**主要処理フロー**:
1. **274行目**: synchronizedブロック開始
2. **275行目**: `!getStatus.getState.isTerminal` で非終端状態を確認
3. **276行目**: `setState(OperationState.FINISHED)` で状態をFINISHEDに変更
4. **277行目**: `HiveThriftServer2.eventManager.onStatementFinish(statementId)` でイベント発行

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 69-72行目: onStatementFinishメソッド。id, System.currentTimeMillis()でイベント生成しpostLiveListenerBusへ委譲 |

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 126行目: onOtherEvent内のSparkListenerThriftServerOperationFinishパターンマッチ |
| 4-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 217-226行目: onOperationFinishedメソッド。finishTimestampとstateをFINISHEDに設定 |

**主要処理フロー**:
- **219行目**: executionData.finishTimestamp = e.finishTime
- **221行目**: executionData.state = ExecutionState.FINISHED
- **222行目**: updateLiveStore(executionData)

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

```
SparkExecuteStatementOperation.execute() [行212-282]
    |
    +-- finally [行273]
            |
            +-- synchronized [行274]
                    |
                    +-- !getStatus.getState.isTerminal [行275]
                    |
                    +-- setState(OperationState.FINISHED) [行276]
                    |
                    +-- HiveThriftServer2.eventManager.onStatementFinish(statementId) [行277]
                            |
                            +-- HiveThriftServer2EventManager.onStatementFinish(id) [行69-72]
                                    |
                                    +-- postLiveListenerBus(SparkListenerThriftServerOperationFinish(id, time)) [行70]
                                            |
                                            +-- sc.listenerBus.post(event) [行29]
                                                    |
                                                    +-- HiveThriftServer2Listener.onOtherEvent(event) [行117]
                                                            |
                                                            +-- onOperationFinished(e) [行126, 行217-226]
                                                                    |
                                                                    +-- executionData.state = ExecutionState.FINISHED [行221]
                                                                    +-- updateLiveStore(executionData) [行222]
```

### データフロー図

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

execute()正常完了         --->  finally synchronized              --->  SparkListenerThriftServerOperationFinish
  |                              |                                       |
  +-- statementId         --->  onStatementFinish(id)             --->  LiveListenerBus.post()
  +-- System.currentTimeMillis()                                         |
                                                                         +---> HiveThriftServer2Listener
                                                                                |
                                                                                +---> ExecutionInfo (KVStore)
                                                                                |     state=FINISHED
                                                                                |     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()のfinallyブロック |
| 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の定義、isExecutionActive |
| HiveThriftServer2ListenerSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2ListenerSuite.scala` | テスト | 完了イベントの処理を含むリスナーテスト（70-71行目） |
| ThriftServerPageSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPageSuite.scala` | テスト | ThriftServer UIページのテスト |
