# 機能設計書 81-Flink CLI

## 概要

本ドキュメントは、Apache FlinkのコマンドラインクライアントであるFlink CLIの機能設計書である。flink-clientsモジュールが提供するCLIツールにより、ユーザーはFlinkジョブの実行、管理、モニタリングをコマンドラインから行うことができる。

### 本機能の処理概要

Flink CLIは、Apache Flinkクラスターに対してジョブの投入・管理を行うためのコマンドラインインターフェースである。ユーザーはシェルからFlinkコマンドを実行し、JARファイルに含まれるFlinkプログラムを実行したり、実行中のジョブの状態を確認・操作したりできる。

**業務上の目的・背景**：本番環境やCI/CDパイプラインからFlinkジョブを自動的に投入・管理する必要がある。GUIベースのWeb UIに加えて、スクリプトやオートメーションツールと連携可能なCLIインターフェースを提供することで、運用の自動化と効率化を実現する。また、開発者がローカル環境でジョブをテスト・デバッグする際にも使用される。

**機能の利用シーン**：(1)開発時のジョブ実行テスト、(2)本番環境へのジョブデプロイ、(3)ジョブの状態確認・モニタリング、(4)ジョブのキャンセル・停止操作、(5)セーブポイント/チェックポイントの管理、(6)CI/CDパイプラインでの自動デプロイ

**主要な処理内容**：
1. **run**: JARファイルからFlinkプログラムをコンパイル・実行
2. **list**: 実行中・スケジュール済み・完了済みのジョブ一覧を表示
3. **info**: プログラムの実行計画（JSON形式）を表示
4. **cancel**: 実行中のジョブをキャンセル
5. **stop**: セーブポイントを取得してジョブを停止（ストリーミングジョブ用）
6. **savepoint**: 手動でセーブポイントをトリガーまたは破棄
7. **checkpoint**: 手動でチェックポイントをトリガー

**関連システム・外部連携**：JobManager REST API経由でクラスターと通信、YARN/Kubernetes等のリソースマネージャーとの連携、セーブポイント/チェックポイント用の分散ファイルシステム（HDFS、S3等）

**権限による制御**：セキュリティ設定が有効な場合、Kerberos認証やSSL/TLS通信が適用される

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | CLIはコマンドラインツールのため関連画面なし |

## 機能種別

コマンドライン操作 / ジョブ管理 / データ連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| action | String | Yes | 実行アクション（run, list, info, cancel, stop, savepoint, checkpoint） | 有効なアクション名であること |
| -j, --jarfile | String | No | Flinkプログラムを含むJARファイルパス | ファイルが存在すること |
| -c, --class | String | No | エントリーポイントクラス名（マニフェストに未指定の場合必須） | 有効なクラス名であること |
| -p, --parallelism | Integer | No | 並列度 | 正の整数であること |
| -d, --detached | Boolean | No | デタッチモードで実行 | - |
| -s, --fromSavepoint | String | No | リストアするセーブポイントのパス | パスが存在すること |
| -m, --jobmanager | String | No | JobManagerのアドレス（host:port） | 有効なホスト:ポート形式であること |
| jobId | String | No(アクションによる) | 操作対象のジョブID | 有効なJobID形式（16進数文字列）であること |

### 入力データソース

- コマンドライン引数
- 設定ファイル（flink-conf.yaml）
- 環境変数（FLINK_CONF_DIR等）
- JARファイル（マニフェスト情報）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| 実行結果メッセージ | String | 各アクションの実行結果を示すテキスト |
| ジョブ一覧 | List | list実行時のジョブ情報（JobID、名前、状態、開始時刻） |
| 実行計画JSON | String | info実行時のジョブ実行計画 |
| セーブポイントパス | String | savepoint/stop成功時のセーブポイント保存先 |
| 終了コード | Integer | 0:成功、1:エラー |

### 出力先

- 標準出力（コンソール）
- 標準エラー出力（エラーメッセージ）
- ログファイル（SLF4J経由）

## 処理フロー

### 処理シーケンス

```
1. mainメソッド実行
   └─ EnvironmentInformation.logEnvironmentInfo()で環境情報ログ出力
2. 設定ディレクトリの特定
   └─ FLINK_CONF_DIR環境変数、またはfallbackディレクトリ（../conf, conf）
3. グローバル設定の読み込み
   └─ GlobalConfiguration.loadConfiguration()
4. カスタムコマンドラインのロード
   └─ GenericCLI、FlinkYarnSessionCli（利用可能な場合）、DefaultCLI
5. セキュリティコンテキストの初期化
   └─ SecurityUtils.install()、SecurityUtils.getInstalledContext().runSecured()
6. parseAndRunの実行
   └─ 引数からアクションを判別し、対応するメソッドを呼び出し
7. アクション固有の処理
   └─ run(): プログラムのビルド・実行
   └─ list(): ジョブ一覧の取得・表示
   └─ cancel(): ジョブのキャンセル
   └─ stop(): セーブポイント取得＋停止
   └─ savepoint(): セーブポイント操作
   └─ checkpoint(): チェックポイント操作
8. クラスターとの通信
   └─ ClusterClientを介してREST APIで通信
9. 結果の出力と終了
   └─ 成功時は終了コード0、エラー時は終了コード1
```

### フローチャート

```mermaid
flowchart TD
    A[main開始] --> B[設定ディレクトリ特定]
    B --> C[グローバル設定読み込み]
    C --> D[カスタムコマンドラインロード]
    D --> E[セキュリティコンテキスト初期化]
    E --> F{アクション判定}
    F -->|run| G[run実行]
    F -->|list| H[list実行]
    F -->|info| I[info実行]
    F -->|cancel| J[cancel実行]
    F -->|stop| K[stop実行]
    F -->|savepoint| L[savepoint実行]
    F -->|checkpoint| M[checkpoint実行]
    F -->|不正| N[エラーメッセージ表示]
    G --> O[クラスターへ接続]
    H --> O
    I --> P[実行計画生成]
    J --> O
    K --> O
    L --> O
    M --> O
    O --> Q[REST API呼び出し]
    Q --> R[結果出力]
    P --> R
    N --> S[終了コード1で終了]
    R --> T{成功?}
    T -->|Yes| U[終了コード0で終了]
    T -->|No| S
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | デタッチモード判定 | runアクションで-dオプション指定時、ジョブ投入後即座に制御を返す | run実行時 |
| BR-002 | セーブポイントリストア | -sオプション指定時、指定パスからステートをリストアして起動 | run実行時 |
| BR-003 | アプリケーションモード判定 | デプロイメントターゲットが"application"で終わる場合、アプリケーションモードで実行 | run実行時 |
| BR-004 | ジョブステータスフィルタリング | listで-r/-s/-aオプションにより表示対象を制御 | list実行時 |
| BR-005 | キャンセル非推奨警告 | cancel -sオプション使用時、stopの使用を推奨するメッセージを表示 | cancel -s実行時 |

### 計算ロジック

特になし

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

### 操作別データベース影響一覧

本機能は直接データベースを操作しない。ジョブ情報はJobManager経由でメモリ上のメタデータとして管理される。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| 1 | CliArgsException | コマンドライン引数が不正 | 正しい引数形式を確認 |
| 1 | ProgramParametrizationException | プログラムパラメータが不正 | プログラムの引数を確認 |
| 1 | ProgramMissingJobException | Flinkジョブがプログラムに含まれていない | execute()呼び出しを確認 |
| 1 | ProgramInvocationException | プログラム実行時のエラー | JARファイル、クラス名を確認 |
| 1 | FlinkException | クラスター接続エラー | JobManagerの稼働状態を確認 |
| 1 | FileNotFoundException | JARファイルが見つからない | ファイルパスを確認 |

### リトライ仕様

REST API呼び出し時、ExponentialWaitStrategyによるリトライが適用される（RestClusterClientConfiguration設定による）

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

CLIツールとしてはトランザクション管理は行わない。ジョブ投入・操作は非同期でJobManagerに委譲される。

## パフォーマンス要件

- CLI起動からアクション完了までの応答時間：通常数秒以内
- REST API呼び出しタイムアウト：ClientOptions.CLIENT_TIMEOUTにより設定可能

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

- Kerberos認証サポート（SecurityConfiguration経由）
- SSL/TLS通信（REST API）
- 設定ファイルから認証情報を読み込む際の保護

## 備考

- Python（PyFlink）プログラムの実行もサポート（--python, --pyFilesオプション）
- YARN、Kubernetesなど各デプロイメント環境に応じたカスタムコマンドラインが動的にロードされる

---

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

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

### 推奨読解順序

#### Step 1: コマンドラインオプションの定義を理解する

まず、CLIで使用可能なオプションの定義を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CliFrontendParser.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java` | 各アクションで使用可能なOptionの定義 |

**主要処理フロー**:
- **43-48行目**: HELP_OPTIONの定義
- **50-51行目**: JAR_OPTIONの定義（-j, --jarfile）
- **71-77行目**: PARALLELISM_OPTIONの定義（-p, --parallelism）
- **415-421行目**: getRunCommandOptions()でrunアクション用オプションを構築
- **655-682行目**: createSavepointRestoreSettings()でセーブポイントリストア設定を生成

**読解のコツ**: Apache Commons CLIのOptionクラスを使用してコマンドライン引数を定義している。各オプションは短縮形（-j）と完全形（--jarfile）の両方を持つ。

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

処理の起点となるCliFrontendクラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CliFrontend.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java` | CLIのメイン処理フロー |

**主要処理フロー**:
- **100-106行目**: アクション名の定数定義（ACTION_RUN, ACTION_LIST等）
- **1320-1327行目**: mainメソッド - エントリーポイント
- **1329-1363行目**: mainInternalメソッド - 設定ロード、セキュリティ初期化
- **1242-1317行目**: parseAndRunメソッド - アクション振り分け
- **175-233行目**: runメソッド - ジョブ実行処理
- **390-426行目**: listメソッド - ジョブ一覧取得
- **603-707行目**: cancelメソッド - ジョブキャンセル

**読解のコツ**: parseAndRunメソッドのswitch文でアクションに応じた処理メソッドを呼び出している。各アクションメソッドは、オプション解析 → クラスター接続 → 操作実行 → 結果出力 の流れで処理を行う。

#### Step 3: プログラムパッケージングを理解する

JARファイルからプログラムを構築する処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PackagedProgram.java | `flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java` | JARファイルからのプログラム構築 |

**主要処理フロー**:
- **119-178行目**: コンストラクタ - JARロード、クラスローダー構築、エントリークラスロード
- **236-243行目**: invokeInteractiveModeForExecution() - mainメソッド呼び出し
- **327-384行目**: callMainMethod() - リフレクションによるmain呼び出し
- **386-456行目**: getEntryPointClassNameFromJar() - マニフェストからエントリークラス取得
- **523-561行目**: extractContainedLibraries() - ネストされたJARの展開

**読解のコツ**: PackagedProgramはBuilderパターンで構築される（645行目以降）。JARファイルのマニフェストからprogram-classまたはMain-Classを読み取り、エントリーポイントを特定する。

#### Step 4: クラスター通信を理解する

ClusterClientを介したクラスターとの通信を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CliFrontend.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java` | runClusterActionメソッド（1183-1210行目） |

**主要処理フロー**:
- **1183-1210行目**: runClusterAction() - ClusterClientを取得してアクションを実行
- **1194-1197行目**: clusterClientFactory.getClusterId()でクラスターID取得
- **1203-1209行目**: clusterDescriptor.retrieve()でClusterClient取得、アクション実行

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

```
CliFrontend.main()
    │
    ├─ mainInternal()
    │      ├─ getConfigurationDirectoryFromEnv()
    │      ├─ GlobalConfiguration.loadConfiguration()
    │      ├─ loadCustomCommandLines()
    │      │      ├─ GenericCLI
    │      │      ├─ FlinkYarnSessionCli（あれば）
    │      │      └─ DefaultCLI
    │      └─ SecurityUtils.install() / runSecured()
    │
    └─ parseAndRun()
           ├─ run()
           │      ├─ getCommandLine()
           │      ├─ validateAndGetActiveCommandLine()
           │      ├─ getEffectiveConfiguration()
           │      ├─ buildProgram() → PackagedProgram
           │      └─ executeProgram() → ClientUtils.executeProgram()
           │
           ├─ list()
           │      └─ runClusterAction()
           │             └─ listJobs() → clusterClient.listJobs()
           │
           ├─ cancel()
           │      └─ runClusterAction()
           │             └─ clusterClient.cancel()
           │
           ├─ stop()
           │      └─ runClusterAction()
           │             └─ stopWithSavepoint() / stopWithDetachedSavepoint()
           │
           ├─ savepoint()
           │      └─ runClusterAction()
           │             └─ triggerSavepoint() / disposeSavepoint()
           │
           └─ checkpoint()
                  └─ runClusterAction()
                         └─ triggerCheckpoint()
```

### データフロー図

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

コマンドライン引数 ───▶ CliFrontendParser.parse() ───▶ CommandLine
                              │
設定ファイル ─────────▶ GlobalConfiguration       ───▶ Configuration
                              │
JARファイル ──────────▶ PackagedProgram           ───▶ Pipeline
                              │
                       ClusterClient              ───▶ REST API呼び出し
                              │
                       JobManager                 ───▶ ジョブ実行/状態変更
                              │
                              └────────────────────────▶ 標準出力/エラー
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CliFrontend.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java` | ソース | CLIメインクラス |
| CliFrontendParser.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java` | ソース | コマンドライン引数パーサー |
| PackagedProgram.java | `flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java` | ソース | JARファイルからのプログラム構築 |
| ProgramOptions.java | `flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java` | ソース | プログラム実行オプション |
| CustomCommandLine.java | `flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java` | ソース | カスタムコマンドラインインターフェース |
| GenericCLI.java | `flink-clients/src/main/java/org/apache/flink/client/cli/GenericCLI.java` | ソース | 汎用CLI実装 |
| DefaultCLI.java | `flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java` | ソース | デフォルトCLI実装 |
| ClientUtils.java | `flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java` | ソース | クライアントユーティリティ |
| RestClusterClient.java | `flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java` | ソース | REST APIクライアント |
| ClientOptions.java | `flink-clients/src/main/java/org/apache/flink/client/cli/ClientOptions.java` | ソース | クライアント設定オプション |
