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

## 概要

本ドキュメントは、Apache Flink のクエリアブルステートクライアント機能に関する機能設計書である。flink-queryable-state-client-java モジュールで提供される、実行中の Flink ジョブのステートを外部からクエリするためのクライアントライブラリについて記載する。

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

### 本機能の処理概要

クエリアブルステートクライアントは、実行中の Flink ストリーミングジョブが保持するキー付きステートに対して、外部アプリケーションからクエリを発行するためのクライアントライブラリである。

**業務上の目的・背景**：
- リアルタイムでジョブのステートを確認（ダッシュボード等）
- ストリーム処理結果の即時参照
- 外部システムからの状態問い合わせ

**機能の利用シーン**：
- リアルタイム集計結果の表示
- ストリーミングジョブのデバッグ
- 外部アプリケーションからの状態確認

**主要な処理内容**：
1. クライアントプロキシへの接続
2. KvStateRequest の送信
3. KvStateResponse の受信・デシリアライズ
4. Immutable State オブジェクトの生成

**関連システム・外部連携**：
- TaskManager 上の Client Proxy
- JobManager（ステート位置解決）
- Netty（ネットワーク通信）

**権限による制御**：特になし（ネットワーク接続による制御）

## 関連画面

本機能はバックエンドライブラリであり、直接関連する画面はない。

## 機能種別

クライアントライブラリ / ステートクエリ

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| remoteHostname | String | Yes | プロキシホスト名 | null不可 |
| remotePort | int | Yes | プロキシポート | 0-65535 |
| jobId | JobID | Yes | ジョブ識別子 | null不可 |
| queryableStateName | String | Yes | クエリアブルステート名 | null不可 |
| key | K | Yes | 検索キー | null不可 |
| keyTypeInfo | TypeInformation<K> | Yes | キーの型情報 | null不可 |
| stateDescriptor | StateDescriptor<S, V> | Yes | ステート記述子 | null不可 |

### 入力データソース

- クライアントアプリケーションからのクエリリクエスト

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| state | CompletableFuture<S extends State> | イミュータブルなステートオブジェクト |

### 出力先

- クライアントアプリケーション

## 処理フロー

### 処理シーケンス

```
1. クライアント初期化
   └─ QueryableStateClient コンストラクタ
   └─ Netty Client 作成
   └─ MessageSerializer 設定

2. ステートクエリ実行
   └─ getKvState() 呼び出し
   └─ キーとネームスペースをシリアライズ
   └─ KvStateRequest 作成
   └─ プロキシに送信

3. レスポンス受信
   └─ KvStateResponse 受信
   └─ Immutable State 生成
   └─ CompletableFuture で返却

4. クライアント終了
   └─ shutdownAndWait() または shutdownAndHandle()
```

### フローチャート

```mermaid
flowchart TD
    A[QueryableStateClient作成] --> B[Client Proxy に接続]
    B --> C[getKvState呼び出し]
    C --> D[Key/Namespace シリアライズ]
    D --> E[KvStateRequest 作成]
    E --> F[sendRequest]
    F --> G[Client Proxy]
    G --> H[JobManager: 位置解決]
    H --> I[TaskManager: ステート取得]
    I --> J[KvStateResponse]
    J --> K[デシリアライズ]
    K --> L{ステート種別}
    L -->|VALUE| M[ImmutableValueState]
    L -->|LIST| N[ImmutableListState]
    L -->|MAP| O[ImmutableMapState]
    L -->|REDUCING| P[ImmutableReducingState]
    L -->|AGGREGATING| Q[ImmutableAggregatingState]
    M --> R[CompletableFuture完了]
    N --> R
    O --> R
    P --> R
    Q --> R
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-75-01 | ポート範囲 | remotePort は 0-65535 の範囲内 | コンストラクタ時 |
| BR-75-02 | イミュータブル | 返却される State は読み取り専用 | 常時 |
| BR-75-03 | サポート状態種別 | VALUE, LIST, MAP, REDUCING, AGGREGATING のみ | 常時 |
| BR-75-04 | 非推奨 | Flink 1.18 以降非推奨 | - |

### 計算ロジック

| ロジックNo | ロジック名 | 内容 |
|-----------|-----------|------|
| CL-75-01 | キーハッシュ | key.hashCode() でキーグループ決定に使用 |

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| UnknownHostException | ホスト解決失敗 | ホスト名が解決できない | ホスト名を確認 |
| UnknownKeyOrNamespaceException | キー不在 | 指定キーが存在しない | キーを確認 |
| UnknownKvStateIdException | ステートID不明 | ステートが見つからない | ステート名を確認 |
| UnknownLocationException | 位置不明 | ステートの位置が解決できない | ジョブ状態を確認 |
| FlinkRuntimeException | ステート種別不対応 | サポートされていないステート種別 | サポート種別を使用 |

### リトライ仕様

クライアントライブラリ自体にはリトライ機能はない。呼び出し側で実装する。

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

本機能はトランザクション管理を行わない。クエリは一貫性のあるスナップショットを返さない場合がある。

## パフォーマンス要件

- 非同期リクエスト（CompletableFuture）
- Netty による高性能ネットワーク通信

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

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

## 備考

- Flink 1.18 で非推奨、将来削除予定
- 代替: Table Store や外部データベースへの出力

---

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

### 推奨読解順序

#### Step 1: クライアント本体を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | QueryableStateClient.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java` | クライアント本体 |

**主要処理フロー（QueryableStateClient.java）**:
- **127-130行目**: コンストラクタ（ホスト名版）
- **139-157行目**: コンストラクタ（InetAddress版）- Client作成
- **168-170行目**: shutdownAndHandle() - 非同期シャットダウン
- **177-184行目**: shutdownAndWait() - 同期シャットダウン
- **226-237行目**: getKvState() - TypeHint版
- **250-265行目**: getKvState() - TypeInformation版
- **279-322行目**: getKvState() - 内部実装（シリアライズ・リクエスト送信）
- **324-339行目**: createState() - Immutable State生成
- **351-366行目**: getKvState() - シリアライズ済みリクエスト送信

#### Step 2: イミュータブルステートを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ImmutableValueState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableValueState.java` | 値ステート |
| 2-2 | ImmutableListState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java` | リストステート |
| 2-3 | ImmutableMapState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableMapState.java` | マップステート |

#### Step 3: ネットワーク通信を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Client.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java` | Nettyクライアント |
| 3-2 | KvStateRequest.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/messages/KvStateRequest.java` | リクエストメッセージ |
| 3-3 | KvStateResponse.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/messages/KvStateResponse.java` | レスポンスメッセージ |

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

```
QueryableStateClient
    │
    ├─ new QueryableStateClient(host, port)
    │      │
    │      └─ new Client<KvStateRequest, KvStateResponse>()
    │
    ├─ getKvState(jobId, stateName, key, keyTypeInfo, descriptor)
    │      │
    │      ├─ KvStateSerializer.serializeKeyAndNamespace()
    │      │
    │      ├─ getKvState(jobId, stateName, keyHashCode, serializedKey)
    │      │      │
    │      │      └─ client.sendRequest(remoteAddress, request)
    │      │
    │      └─ createState(response, descriptor)
    │             │
    │             └─ STATE_FACTORIES.get(type).createState()
    │                    │
    │                    └─ ImmutableValueState / ImmutableListState / ...
    │
    └─ shutdownAndWait()
           │
           └─ client.shutdown().get()
```

### データフロー図

```
[クライアントアプリ]         [QueryableStateClient]        [Client Proxy]       [TaskManager]

     │                              │                          │                    │
     │   getKvState()               │                          │                    │
     └─────────────────────────────▶│                          │                    │
                                    │                          │                    │
                        serializeKeyAndNamespace               │                    │
                                    │                          │                    │
                          KvStateRequest                       │                    │
                                    │                          │                    │
                        client.sendRequest()                   │                    │
                                    │                          │                    │
                                    └─────────────────────────▶│                    │
                                                               │                    │
                                                    位置解決リクエスト              │
                                                     (JobManager経由)              │
                                                               │                    │
                                                               └───────────────────▶│
                                                                                    │
                                                                          ステート読み取り
                                                                                    │
                                                               ◀────────────────────┘
                                                               │
                                                    KvStateResponse
                                    ◀──────────────────────────┘
                                    │
                           createState()
                                    │
                        ImmutableValueState等
     ◀──────────────────────────────┘
     │
CompletableFuture<State>
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| QueryableStateClient.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java` | ソース | クライアント本体 |
| Client.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java` | ソース | Nettyクライアント |
| KvStateRequest.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/messages/KvStateRequest.java` | ソース | リクエスト |
| KvStateResponse.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/messages/KvStateResponse.java` | ソース | レスポンス |
| ImmutableValueState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableValueState.java` | ソース | 値ステート |
| ImmutableListState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java` | ソース | リストステート |
| ImmutableMapState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableMapState.java` | ソース | マップステート |
| ImmutableReducingState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableReducingState.java` | ソース | リデューシングステート |
| ImmutableAggregatingState.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableAggregatingState.java` | ソース | 集約ステート |
| KvStateSerializer.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/serialization/KvStateSerializer.java` | ソース | シリアライザ |
| MessageSerializer.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java` | ソース | メッセージシリアライザ |
| UnknownKeyOrNamespaceException.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/exceptions/UnknownKeyOrNamespaceException.java` | ソース | 例外 |
| UnknownKvStateIdException.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/exceptions/UnknownKvStateIdException.java` | ソース | 例外 |
| KvStateID.java | `flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/KvStateID.java` | ソース | ステートID |
