# 通知設計書 22-notifyKvStateUnregistered

## 概要

本ドキュメントは、Apache Flink の KvState 登録解除通知（notifyKvStateUnregistered）の詳細設計を記載したものである。KvState インスタンスがタスクマネージャから登録解除された際に、JobMaster へ通知を送信する仕組みを定義する。

### 本通知の処理概要

**業務上の目的・背景**：Queryable State 機能において、オペレータが停止したり、状態が不要になった場合に、その KvState の登録を解除する必要がある。登録解除時に JobMaster に通知することで、位置情報レジストリから当該状態を削除し、外部クライアントが存在しない状態にクエリを送信することを防ぐ。これにより、分散システムにおける状態の一貫性を維持する。

**通知の送信タイミング**：KvState インスタンスが `KvStateRegistry.unregisterKvState()` メソッドによって登録解除される際に発生する。具体的には、オペレータが停止する際や、タスクの実行終了時に呼び出される。

**通知の受信者**：JobMaster（`KvStateRegistryGateway` を実装するコンポーネント）が受信者となる。`RpcKvStateRegistryListener` が中間層として RPC 経由で JobMaster に通知を転送する。

**通知内容の概要**：通知には、ジョブID、ジョブ頂点ID、キーグループ範囲、登録名が含まれる。登録時と異なり、KvStateID や サーバアドレスは含まれない（解除対象は他のパラメータで一意に特定可能）。

**期待されるアクション**：JobMaster は受信した情報を基に `KvStateLocationRegistry` から当該状態の位置情報を削除し、以降のクエリで当該状態が発見されないようにする。

## 通知種別

内部 RPC 通知（Flink RPC フレームワークによるプロセス間通信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（RPC） |
| 優先度 | 中 |
| リトライ | なし（RPC フレームワーク依存） |

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

`KvStateRegistry` に登録されている `KvStateRegistryListener` に対して通知が送信される。リスナーは `JobID` をキーとして登録されており、対応するジョブのリスナーが呼び出される。登録時と同様に、`HighAvailabilityServices.DEFAULT_JOB_ID` で登録されたレガシーリスナーも優先的にチェックされる。

## 通知テンプレート

### RPC 通知の場合

| 項目 | 内容 |
|-----|------|
| インターフェース | `KvStateRegistryGateway` |
| メソッド | `notifyKvStateUnregistered` |
| 形式 | RPC メソッド呼び出し |

### パラメータ

```java
void notifyKvStateUnregistered(
    JobID jobId,
    JobVertexID jobVertexId,
    KeyGroupRange keyGroupRange,
    String registrationName
);
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| jobId | ジョブ識別子 | 実行コンテキスト | Yes |
| jobVertexId | ジョブ頂点識別子 | タスク情報 | Yes |
| keyGroupRange | キーグループ範囲 | 状態バックエンド | Yes |
| registrationName | 登録名 | 状態記述子 | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 状態解除 | `unregisterKvState()` 呼び出し | 登録済み状態が存在 | オペレータ停止時等の状態解除 |
| タスク終了 | タスク実行終了 | 登録済み状態が存在 | タスクライフサイクル終了 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 状態未登録 | 解除対象の KvStateID が `registeredKvStates` に存在しない場合 |
| リスナー未登録 | `KvStateRegistryListener` が未登録の場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[KvState 登録解除要求] --> B[KvStateRegistry.unregisterKvState]
    B --> C{登録済み?}
    C -->|No| D[処理終了]
    C -->|Yes| E[registeredKvStates から削除]
    E --> F[entry.clear 呼び出し]
    F --> G{リスナー登録済み?}
    G -->|Yes| H[listener.notifyKvStateUnregistered]
    G -->|No| I[通知スキップ]
    H --> J[RpcKvStateRegistryListener]
    J --> K[RPC経由でJobMasterに通知]
    K --> L[KvStateLocationRegistry更新]
    L --> M[終了]
    I --> M
    D --> M
```

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

### 参照テーブル一覧

該当なし（インメモリ管理）

### メモリ内データ構造

| データ構造 | クラス | 用途 | 備考 |
|-----------|------|------|------|
| registeredKvStates | `ConcurrentHashMap<KvStateID, KvStateEntry>` | 登録済み状態の管理 | KvStateRegistry |
| listeners | `ConcurrentHashMap<JobID, KvStateRegistryListener>` | リスナー管理 | KvStateRegistry |

### 更新データ一覧

| データ構造 | 操作 | 概要 |
|-----------|------|------|
| registeredKvStates | REMOVE | KvState の登録解除 |
| KvStateLocationRegistry | 削除 | JobMaster 側での位置情報削除 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 状態未登録 | 解除対象が存在しない | 何もせず正常終了 |
| RPC失敗 | ネットワーク障害等 | RPC フレームワークに委任 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | RPC フレームワーク依存 |
| リトライ間隔 | RPC フレームワーク依存 |
| リトライ対象エラー | ネットワーク一時障害 |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし（システム稼働中常時）

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

- RPC 通信は Flink の認証・暗号化設定に従う
- 登録解除通知は正当な送信元からのみ受け付ける必要がある

## 備考

- 本通知は Queryable State 機能が有効な場合にのみ使用される
- `notifyKvStateRegistered` と対になる通知であり、状態ライフサイクルの終了を表す
- 解除時には `entry.clear()` が呼び出され、リソースがクリーンアップされる

---

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

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

### 推奨読解順序

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

解除に必要なパラメータと、登録時と同じデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KvStateID.java | `flink-queryable-state/flink-queryable-state-common/src/main/java/org/apache/flink/queryablestate/KvStateID.java` | 解除時のキーとして使用 |
| 1-2 | KvStateEntry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateEntry.java` | clear() メソッドによるクリーンアップ |

**読解のコツ**: 登録解除では `KvStateID` をキーとして `registeredKvStates` マップから削除し、`KvStateEntry.clear()` でリソースを解放する。

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

登録解除のメインロジックを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KvStateRegistry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java` | 状態解除とリスナー通知のメインロジック |

**主要処理フロー**:
1. **行 122-139**: `unregisterKvState()` メソッド - 解除のエントリーポイント
2. **行 129**: `registeredKvStates.remove()` による削除
3. **行 130-131**: 削除成功時に `entry.clear()` 呼び出し
4. **行 133-137**: リスナーへの通知呼び出し

#### Step 3: リスナーインターフェースを理解する

通知を受け取るリスナーのインターフェース定義を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | KvStateRegistryListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistryListener.java` | リスナーインターフェース定義 |

**主要処理フロー**:
- **行 57-61**: `notifyKvStateUnregistered()` メソッドシグネチャ（KvStateID は含まれない点に注意）

#### Step 4: RPC 実装を理解する

実際の RPC 通信を行う実装クラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | RpcKvStateRegistryListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcKvStateRegistryListener.java` | RPC 通信の実装 |

**主要処理フロー**:
- **行 59-68**: `notifyKvStateUnregistered()` の実装 - Gateway 経由で JobMaster に通知

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

```
TaskManager (オペレータ終了)
    │
    ├─ AbstractStreamOperator.dispose() / タスク終了
    │      └─ TaskKvStateRegistry.unregisterKvState()
    │              └─ KvStateRegistry.unregisterKvState()  [行 122-139]
    │                      │
    │                      ├─ registeredKvStates.remove()  [行 129]
    │                      ├─ entry.clear()  [行 131]
    │                      └─ listener.notifyKvStateUnregistered()  [行 135-136]
    │                              └─ RpcKvStateRegistryListener.notifyKvStateUnregistered()
    │                                      └─ KvStateRegistryGateway.notifyKvStateUnregistered() [RPC]
    │
    └─ JobMaster (受信側)
            └─ KvStateLocationRegistry.notifyKvStateUnregistered()
```

### データフロー図

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

KvStateID           ───▶  KvStateRegistry           ───▶  RPC 通知
(解除対象の識別子)           (削除・クリーンアップ)           (JobMaster へ)
      │                          │                            │
      │                          ▼                            ▼
      │               registeredKvStates Map          KvStateLocationRegistry
      │               (エントリ削除)                   (位置情報削除)
      │                          │
      │                          ▼
      │                   entry.clear()
      │                   (リソース解放)
      │                          │
      └──────────────────────────┘
              状態メタデータ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KvStateRegistry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/` | ソース | 状態登録管理 |
| KvStateRegistryListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/` | ソース | リスナーインターフェース |
| RpcKvStateRegistryListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/` | ソース | RPC 実装 |
| KvStateRegistryGateway.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/` | ソース | RPC ゲートウェイ |
| KvStateLocationRegistry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/` | ソース | 位置情報管理 |
| TaskKvStateRegistry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/` | ソース | タスク用ファサード |
| KvStateEntry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/` | ソース | 状態エントリ（clear メソッド） |
