# 機能設計書 76-クエリアブルステートランタイム

## 概要

本ドキュメントは、Apache Flink のクエリアブルステートランタイム機能に関する機能設計書である。flink-queryable-state-runtime モジュールで提供される、TaskManager 上でステートクエリを処理するサーバーコンポーネントについて記載する。

> **非推奨警告**: この機能は Flink 1.18 以降非推奨となり、将来のメジャーバージョンで削除される予定である。

### 本機能の処理概要

クエリアブルステートランタイムは、TaskManager 上で動作する 2 つのサーバーコンポーネントで構成される：
1. **KvStateServer** - 実際のステートを保持し、クエリに応答するサーバー
2. **KvStateClientProxy** - クライアントからのリクエストを受け付け、適切な KvStateServer に転送するプロキシ

**業務上の目的・背景**：
- 外部クライアントからの分散ステートへのアクセスを可能にする
- ステートの位置解決とリクエストルーティングの実装
- クライアントとステート間の通信基盤提供

**機能の利用シーン**：
- QueryableStateClient からのクエリ処理
- ステート位置の動的解決
- 分散ステートへの透過的アクセス

**主要な処理内容**：
1. KvStateClientProxy でのクライアントリクエスト受信
2. JobManager 経由でのステート位置解決
3. KvStateServer へのリクエスト転送
4. ステート読み取りとレスポンス返却

**関連システム・外部連携**：
- QueryableStateClient（クライアント側）
- JobManager（ステート位置解決）
- KvStateRegistry（ステート登録）
- Netty（ネットワーク通信）

**権限による制御**：特になし

## 関連画面

本機能はバックエンドサーバーコンポーネントであり、直接関連する画面はない。

## 機能種別

サーバーランタイム / ステートクエリ処理

## 入力仕様

### 入力パラメータ（KvStateServerImpl）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| bindAddress | String | Yes | バインドアドレス | null不可 |
| bindPortIterator | Iterator<Integer> | Yes | バインドポート範囲 | - |
| numEventLoopThreads | Integer | Yes | イベントループスレッド数 | - |
| numQueryThreads | Integer | Yes | クエリ処理スレッド数 | - |
| kvStateRegistry | KvStateRegistry | Yes | ステートレジストリ | null不可 |
| stats | KvStateRequestStats | Yes | 統計コレクター | null不可 |

### 入力パラメータ（KvStateClientProxyImpl）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| bindAddress | String | Yes | バインドアドレス | null不可 |
| bindPortIterator | Iterator<Integer> | Yes | バインドポート範囲 | - |
| numEventLoopThreads | Integer | Yes | イベントループスレッド数 | - |
| numQueryThreads | Integer | Yes | クエリ処理スレッド数 | >= 1 |
| stats | KvStateRequestStats | Yes | 統計コレクター | null不可 |

### 入力データソース

- KvStateRequest（クライアントから）
- KvStateInternalRequest（プロキシから）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| KvStateResponse | KvStateResponse | シリアライズされたステートデータ |

### 出力先

- クライアントアプリケーション（プロキシ経由）

## 処理フロー

### 処理シーケンス

```
1. サーバー起動
   └─ KvStateClientProxyImpl.start()
   └─ KvStateServerImpl.start()
   └─ Netty サーバー初期化

2. ステート登録
   └─ KvStateRegistry にステート登録
   └─ JobManager に位置を報告

3. クエリ処理（プロキシ）
   └─ KvStateRequest 受信
   └─ getKvStateLocationOracle() で位置オラクル取得
   └─ requestKvStateLocation() でステート位置解決
   └─ KvStateInternalRequest を KvStateServer に送信

4. クエリ処理（サーバー）
   └─ KvStateInternalRequest 受信
   └─ KvStateRegistry からステート取得
   └─ ステートデータ読み取り
   └─ KvStateResponse 返却

5. サーバー停止
   └─ shutdown() 呼び出し
   └─ shutdownServer().get() で同期待機
```

### フローチャート

```mermaid
flowchart TD
    subgraph Client [Client]
        A[QueryableStateClient]
    end

    subgraph Proxy [KvStateClientProxy]
        B[KvStateClientProxyImpl]
        C[KvStateClientProxyHandler]
    end

    subgraph Server [KvStateServer]
        D[KvStateServerImpl]
        E[KvStateServerHandler]
        F[KvStateRegistry]
    end

    subgraph JM [JobManager]
        G[KvStateLocationOracle]
    end

    A -->|KvStateRequest| B
    B --> C
    C -->|位置解決| G
    G -->|KvStateLocation| C
    C -->|KvStateInternalRequest| D
    D --> E
    E -->|lookup| F
    F -->|State| E
    E -->|KvStateResponse| C
    C -->|KvStateResponse| A
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-76-01 | クエリスレッド数 | numQueryThreads >= 1 | コンストラクタ時 |
| BR-76-02 | シャットダウンタイムアウト | 10秒でシャットダウン完了待機 | shutdown()時 |
| BR-76-03 | デフォルトジョブID優先 | レガシーコード互換のため DEFAULT_JOB_ID のオラクルを優先 | 位置解決時 |
| BR-76-04 | リフレクション生成 | QueryableStateUtils経由でリフレクションにより生成 | サーバー生成時 |

### 計算ロジック

特になし

## データベース操作仕様

本機能はデータベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| UnknownKvStateIdException | ステートID不明 | レジストリにステートがない | ジョブ状態を確認 |
| UnknownKeyOrNamespaceException | キー不在 | 指定キーが存在しない | キーを確認 |
| UnknownLocationException | 位置不明 | ステート位置が解決できない | ジョブ状態を確認 |
| BindException | ポートバインド失敗 | 指定ポートが使用中 | ポート設定を確認 |

### リトライ仕様

サーバー側での自動リトライはない。クライアント側で実装する。

## トランザクション仕様

本機能はトランザクション管理を行わない。

## パフォーマンス要件

- Netty による非同期 I/O
- スレッドプール設定による並列処理

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

- ネットワーク接続による認証なし（ファイアウォールで保護推奨）

## 備考

- Flink 1.18 で非推奨、将来削除予定

---

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

### 推奨読解順序

#### Step 1: サーバー実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KvStateServerImpl.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java` | ステートサーバー実装 |
| 1-2 | KvStateServerHandler.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java` | リクエストハンドラ |

**主要処理フロー（KvStateServerImpl.java）**:
- **67-83行目**: コンストラクタ - サーバー初期化
- **86-92行目**: initializeHandler() - ハンドラ初期化
- **101-103行目**: start() - サーバー起動
- **111-118行目**: shutdown() - サーバー停止

#### Step 2: プロキシ実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KvStateClientProxyImpl.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java` | クライアントプロキシ実装 |
| 2-2 | KvStateClientProxyHandler.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java` | プロキシハンドラ |

**主要処理フロー（KvStateClientProxyImpl.java）**:
- **71-89行目**: コンストラクタ - プロキシ初期化
- **97-99行目**: start() - プロキシ起動
- **102-109行目**: shutdown() - プロキシ停止
- **112-119行目**: updateKvStateLocationOracle() - 位置オラクル更新
- **122-134行目**: getKvStateLocationOracle() - 位置オラクル取得（DEFAULT_JOB_ID優先）
- **137-143行目**: initializeHandler() - ハンドラ初期化

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

```
KvStateClientProxyImpl (Client Proxy)
    │
    ├─ start()
    │      └─ AbstractServerBase.start()
    │             └─ Netty Bootstrap
    │
    ├─ initializeHandler()
    │      └─ new KvStateClientProxyHandler()
    │
    └─ KvStateClientProxyHandler.handleRequest()
           │
           ├─ getKvStateLocationOracle(jobId)
           │
           ├─ kvStateLocationOracle.requestKvStateLocation()
           │      └─ JobManager に位置問い合わせ
           │
           └─ KvStateInternalRequest を KvStateServer に送信

KvStateServerImpl (State Server)
    │
    ├─ start()
    │      └─ AbstractServerBase.start()
    │
    ├─ initializeHandler()
    │      └─ new KvStateServerHandler()
    │
    └─ KvStateServerHandler.handleRequest()
           │
           ├─ kvStateRegistry.getKvState()
           │
           └─ KvStateResponse 生成
```

### データフロー図

```
[QueryableStateClient]       [KvStateClientProxy]        [JobManager]        [KvStateServer]

        │                           │                          │                    │
        │   KvStateRequest          │                          │                    │
        └──────────────────────────▶│                          │                    │
                                    │                          │                    │
                             handleRequest()                   │                    │
                                    │                          │                    │
                    getKvStateLocationOracle()                 │                    │
                                    │                          │                    │
                     requestKvStateLocation()                  │                    │
                                    └─────────────────────────▶│                    │
                                                               │                    │
                                                    KvStateLocation               │
                                    ◀──────────────────────────┘                    │
                                    │                                               │
                         KvStateInternalRequest                                     │
                                    └───────────────────────────────────────────────▶
                                                                                    │
                                                                 kvStateRegistry.getKvState()
                                                                                    │
                                                                          State読み取り
                                                                                    │
                                                               KvStateResponse      │
                                    ◀────────────────────────────────────────────────
                                    │
        KvStateResponse             │
        ◀───────────────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KvStateServerImpl.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java` | ソース | ステートサーバー |
| KvStateServerHandler.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java` | ソース | サーバーハンドラ |
| KvStateClientProxyImpl.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java` | ソース | クライアントプロキシ |
| KvStateClientProxyHandler.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java` | ソース | プロキシハンドラ |
| KvStateInternalRequest.java | `flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/messages/KvStateInternalRequest.java` | ソース | 内部リクエスト |
| AbstractServerBase.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java` | ソース | サーバー基底 |
| AbstractServerHandler.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java` | ソース | ハンドラ基底 |
| KvStateRegistry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java` | ソース | ステートレジストリ |
| KvStateLocationOracle.java | `flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/KvStateLocationOracle.java` | ソース | 位置オラクル |
