# 通知設計書 104-SparkListenerThriftServerOperationClosed

## 概要

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

### 本通知の処理概要

ThriftServerオペレーションのクローズを検知し、対応するリスナーへ通知するイベント機構である。SparkOperationトレイトのclose()メソッド内で、親クラスのclose()実行後にHiveThriftServer2EventManager.onOperationClosed()を通じてイベントが生成される。このイベントはオペレーションのライフサイクルにおける最終的な終了通知であり、受信側ではオペレーション状態をCLOSEDに遷移させ、ライブ管理データ（executionList）から削除し、KVStoreに最終状態を永続化する。

**業務上の目的・背景**：HiveThriftServer2においてオペレーションのクローズはリソース解放の完了を意味する。本通知は、オペレーションが完全にクローズされた事実をシステム全体に伝搬し、(1) ThriftServer UIのランニングカウントから除外する、(2) KVStoreにクローズ時刻を記録し実行履歴として永続化する、(3) executionListからライブデータを削除しメモリを解放する、という3つの目的を達成する。

**通知の送信タイミング**：SparkOperationトレイトのclose()メソッド内で、親クラスのOperation.close()呼び出し後、cleanup()実行後にeventManager.onOperationClosed()が呼び出される。close()はクライアントがオペレーション結果の取得を完了した後、またはセッションクローズ時に呼び出される。

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

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

**期待されるアクション**：受信者はオペレーションの状態をCLOSEDに更新し、closeTimestampを設定する。updateStoreWithTriggerEnabled()により履歴ストアに永続化した後、executionListからデータを削除する。これによりオペレーションはUI上で完全にクローズされた状態として表示され、アクティブオペレーションカウントから除外される。

## 通知種別

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

## 送信仕様

### 基本情報

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

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

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

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```
SparkListenerThriftServerOperationClosed(id={オペレーションID}, closeTime={クローズ時刻（ミリ秒エポック）})
```

### 添付ファイル

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

## テンプレート変数

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

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| オペレーションクローズ | SparkOperation.close() | 常時（close()が呼ばれた時点で必ず発火） | 親クラスのclose()とcleanup()実行後にイベント発行（SparkOperation.scala 行65-70） |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| なし | close()が呼ばれた場合は常にイベントが発火する。close()自体の呼び出しはHiveのOperation基盤クラスで管理される |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[クライアントがオペレーションをクローズ] --> B[SparkOperation.close]
    B --> C[super.close - 親クラスのclose]
    C --> D[cleanup実行]
    D --> E[eventManager.onOperationClosed]
    E --> F[SparkListenerThriftServerOperationClosed生成]
    F --> G[LiveListenerBus.post]
    G --> H[HiveThriftServer2Listener.onOtherEvent]
    H --> I[onOperationClosed]
    I --> J{executionListにオペレーションIDが存在?}
    J -->|Yes| K[closeTimestamp設定]
    K --> L[state = CLOSED]
    L --> M[updateStoreWithTriggerEnabled - KVStore永続化]
    M --> N[executionList.remove - ライブデータ削除]
    J -->|No| O[logWarning出力]
```

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

### 参照テーブル一覧

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

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

#### executionList（ConcurrentHashMap）

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

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| KVStore（ElementTrackingStore） | UPDATE | ExecutionInfoのcloseTimestampとstateを更新（トリガー有効） |
| executionList（ConcurrentHashMap） | DELETE | ライブデータをインメモリマップから削除 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | closeTimestamp | e.closeTime | クローズ時刻 |
| UPDATE | state | ExecutionState.CLOSED | クローズ状態 |
| DELETE | executionList[id] | - | ライブデータの削除 |

## エラー処理

### エラーケース一覧

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

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

配信時間帯の制限なし。オペレーションがクローズされた時点で即時配信される。

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

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

## 備考

- OperationClosedはオペレーションライフサイクルの最終イベントである。通常のフローでは、OperationStart -> OperationParsed -> OperationFinish -> OperationClosed の順でイベントが発火する
- 他の終端状態（CANCELED, TIMEDOUT, ERROR）のオペレーションも最終的にclose()が呼ばれるため、OperationClosedイベントが発火する
- onOperationClosedではupdateLiveStore()ではなくupdateStoreWithTriggerEnabled()が使用される。これはクローズ時にトリガーを有効化し、retainedStatements閾値を超えた古いエントリのクリーンアップを行うためである
- executionList.remove(e.id)によりライブデータがメモリから解放されるため、リスナーのnoLiveData()の結果に影響する

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HiveThriftServer2EventManager.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala` | 116-118行目: SparkListenerThriftServerOperationClosedケースクラスの定義（id, closeTimeフィールド）。finishTimeではなくcloseTimeである点に注意 |
| 1-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 316-346行目: LiveExecutionDataクラス。closeTimestampフィールド（行324）がOperationClosed専用 |
| 1-3 | HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | 117-123行目: isExecutionActive定義。CLOSEDは非アクティブ |

**読解のコツ**: OperationFinishイベントではfinishTimestampが更新され、OperationClosedイベントではcloseTimestampが更新される。この2つのタイムスタンプは独立しており、OperationFinishからOperationClosedまでの時間が結果取得にかかった時間を示す。

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

オペレーションクローズの起点であるSparkOperationトレイトを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala` | 65-70行目: close()メソッド。abstract overrideでsuper.close()後にcleanup()とonOperationClosed()を呼び出す |

**主要処理フロー**:
1. **66行目**: `super.close()` - 親クラスのOperation.close()を呼び出し
2. **67行目**: `cleanup()` - リソースのクリーンアップ
3. **68行目**: `logInfo` - ログ出力
4. **69行目**: `HiveThriftServer2.eventManager.onOperationClosed(statementId)` - イベント発行

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

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

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

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

HiveThriftServer2Listenerによるイベント受信、状態更新、ライブデータ削除を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 127行目: onOtherEvent内のSparkListenerThriftServerOperationClosedパターンマッチ |
| 4-2 | HiveThriftServer2Listener.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala` | 228-238行目: onOperationClosedメソッド。closeTimestamp設定、CLOSED状態遷移、updateStoreWithTriggerEnabled、executionList.remove |

**主要処理フロー**:
- **231行目**: executionData.closeTimestamp = e.closeTime
- **232行目**: executionData.state = ExecutionState.CLOSED
- **233行目**: updateStoreWithTriggerEnabled(executionData) - トリガー有効でKVStore更新
- **234行目**: executionList.remove(e.id) - ライブデータ削除

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

```
SparkOperation.close() [行65-70]
    |
    +-- super.close() [行66]
    |
    +-- cleanup() [行67]
    |
    +-- HiveThriftServer2.eventManager.onOperationClosed(statementId) [行69]
            |
            +-- HiveThriftServer2EventManager.onOperationClosed(id) [行74-76]
                    |
                    +-- postLiveListenerBus(SparkListenerThriftServerOperationClosed(id, time)) [行75]
                            |
                            +-- sc.listenerBus.post(event) [行29]
                                    |
                                    +-- HiveThriftServer2Listener.onOtherEvent(event) [行117]
                                            |
                                            +-- onOperationClosed(e) [行127, 行228-238]
                                                    |
                                                    +-- executionData.closeTimestamp = e.closeTime [行231]
                                                    +-- executionData.state = ExecutionState.CLOSED [行232]
                                                    +-- updateStoreWithTriggerEnabled(executionData) [行233]
                                                    +-- executionList.remove(e.id) [行234]
```

### データフロー図

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

SparkOperation.close()    --->  onOperationClosed(id)            --->  SparkListenerThriftServerOperationClosed
  |                              |                                       |
  +-- statementId         --->  イベント生成                       --->  LiveListenerBus.post()
  +-- System.currentTimeMillis()                                         |
                                                                         +---> HiveThriftServer2Listener
                                                                                |
                                                                                +---> ExecutionInfo (KVStore)
                                                                                |     state=CLOSED
                                                                                |     closeTimestamp=closeTime
                                                                                |
                                                                                +---> executionList.remove (メモリ解放)
                                                                                |
                                                                                +---> 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永続化・ライブデータ削除 |
| SparkOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala` | ソース | close()メソッドのエントリーポイント。abstract overrideによる拡張 |
| HiveThriftServer2.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala` | ソース | eventManagerの初期化。ExecutionState列挙体の定義（CLOSED含む） |
| HiveThriftServer2AppStatusStore.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2AppStatusStore.scala` | ソース | KVStore経由のステータス参照。ExecutionInfoの定義、isExecutionActive |
| SparkExecuteStatementOperation.scala | `sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala` | ソース | cleanup()の具体実装（backgroundHandle.cancel, cancelJobGroup） |
| HiveThriftServer2ListenerSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2ListenerSuite.scala` | テスト | クローズイベントの処理を含むリスナーテスト（72-73行目, 139-140行目） |
| ThriftServerPageSuite.scala | `sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPageSuite.scala` | テスト | ThriftServer UIページのテスト |
