# 機能設計書 71-RPCシステム

## 概要

本ドキュメントは、Apache Flink の RPC（Remote Procedure Call）システムに関する機能設計書である。flink-rpc モジュールで提供されるコンポーネント間通信基盤について、その構造、処理フロー、および実装詳細を記載する。

### 本機能の処理概要

RPCシステムは、Flink クラスターを構成する分散コンポーネント（JobManager、TaskManager、ResourceManager など）間の非同期通信を実現する基盤機能である。

**業務上の目的・背景**：分散システムにおいて、複数のプロセスやノード間でメソッド呼び出しを透過的に行う必要がある。Flinkでは、JobManagerがジョブの管理を行い、TaskManagerがタスクを実行するという分散アーキテクチャを採用しているため、これらのコンポーネント間で効率的かつ信頼性の高い通信手段が不可欠である。RPCシステムは、この通信を抽象化し、開発者がローカルメソッド呼び出しと同様のインターフェースでリモート通信を行えるようにする。

**機能の利用シーン**：
- JobManager と TaskManager 間のタスク割り当て・状態報告
- ResourceManager と TaskManager 間のリソース管理通信
- チェックポイント調整のための分散コーディネーション
- ハートビート監視による障害検出

**主要な処理内容**：
1. RpcEndpoint の登録とサーバー起動
2. RpcGateway を通じたリモートメソッド呼び出しのプロキシ生成
3. メッセージのシリアライズ・デシリアライズと送受信
4. 非同期レスポンスの CompletableFuture による管理
5. フェンシングトークンによる古いリーダーからのメッセージ拒否

**関連システム・外部連携**：
- Apache Pekko（旧 Akka）アクターシステム（デフォルト実装）
- Netty ネットワーク層

**権限による制御**：特になし（システム内部の通信基盤のため）

## 関連画面

本機能はバックエンドの通信基盤であり、直接関連する画面はない。

## 機能種別

通信基盤 / システム内部連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| address | String | Yes | 接続先RPCエンドポイントのアドレス | 有効なRPC URL形式 |
| fencingToken | Serializable | No | フェンシング用トークン | Serializableであること |
| timeout | Duration | No | RPC呼び出しのタイムアウト | 正の値 |

### 入力データソース

- RpcEndpoint を継承したコンポーネント（JobMaster、ResourceManager、TaskExecutor など）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| RpcGateway | RpcGateway実装 | リモート呼び出し用プロキシ |
| CompletableFuture | CompletableFuture<T> | 非同期RPC呼び出しの結果 |
| terminationFuture | CompletableFuture<Void> | エンドポイント終了の完了通知 |

### 出力先

- 呼び出し元コンポーネント（CompletableFuture として返却）

## 処理フロー

### 処理シーケンス

```
1. RpcEndpoint の初期化
   └─ RpcService.startServer() により Actor を生成しサーバー起動

2. RpcGateway の取得（リモート接続）
   └─ RpcService.connect() により指定アドレスに接続
   └─ ハンドシェイクでバージョン確認
   └─ InvocationHandler を設定したプロキシを返却

3. RPC メソッド呼び出し
   └─ プロキシの invoke() が RpcInvocation メッセージを生成
   └─ Pekko Actor にメッセージ送信
   └─ 受信側 Actor がメソッドを反射呼び出し
   └─ 結果を送信元に返却

4. RpcEndpoint の終了
   └─ closeAsync() で停止処理開始
   └─ onStop() コールバック実行
   └─ Actor 停止、リソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[RpcEndpoint 初期化] --> B[RpcService.startServer]
    B --> C{サーバー起動成功?}
    C -->|Yes| D[Actor 生成・登録]
    C -->|No| E[例外スロー]
    D --> F[start により STARTED 状態へ]
    F --> G[RPC 呼び出し待機]
    G --> H{メッセージ受信}
    H -->|RpcInvocation| I[メソッド呼び出し実行]
    H -->|RunAsync| J[Runnable 実行]
    H -->|CallAsync| K[Callable 実行]
    I --> L[結果返却]
    J --> G
    K --> L
    L --> G
    G --> M{終了要求?}
    M -->|Yes| N[onStop 呼び出し]
    N --> O[Actor 停止]
    O --> P[terminationFuture 完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-71-01 | シングルスレッド実行 | 同一エンドポイントへのRPC呼び出しは単一スレッドで実行 | 常時 |
| BR-71-02 | フェンシング検証 | FencedRpcEndpoint の場合、トークン一致時のみ処理 | FencedRpcGateway 使用時 |
| BR-71-03 | ハンドシェイク必須 | 接続時にバージョン互換性をハンドシェイクで確認 | リモート接続時 |
| BR-71-04 | タイムアウト処理 | 指定時間内に応答がない場合は TimeoutException | 常時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RpcConnectionException | 接続エラー | 指定アドレスに接続できない | アドレス確認、ネットワーク確認 |
| HandshakeException | ハンドシェイクエラー | バージョン不一致、Gateway未サポート | Flinkバージョンの統一 |
| FencingTokenException | フェンシングエラー | トークン不一致 | 正しいリーダーに接続し直す |
| TimeoutException | タイムアウト | 応答時間超過 | タイムアウト値調整、負荷確認 |
| RecipientUnreachableException | 到達不能 | 受信側エンドポイントが終了 | 再接続処理 |

### リトライ仕様

RPC呼び出しレベルでのリトライは組み込まれていない。呼び出し側で必要に応じてリトライロジックを実装する。

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

RPCシステム自体はトランザクションを管理しない。必要な場合は呼び出し側で管理する。

## パフォーマンス要件

- ローカル呼び出し: マイクロ秒オーダー
- リモート呼び出し: ネットワーク遅延 + シリアライゼーション時間
- デフォルトタイムアウト: 設定による（通常10秒程度）

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

- SSL/TLS による通信暗号化をサポート（CustomSSLEngineProvider）
- 認証機構は Pekko の設定で構成可能

## 備考

- RPCシステムは Pekko ベースの実装がデフォルト
- SPI（ServiceLoader）を通じて実装を切り替え可能

---

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

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

### 推奨読解順序

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

まず、RPC通信で使用されるメッセージ型とインターフェースを理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | RpcGateway.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java` | RPCインターフェースの基底。getAddress()、getHostname() |
| 1-2 | RpcInvocation.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/messages/RpcInvocation.java` | RPCメッセージの構造。メソッド名、パラメータ型、引数 |
| 1-3 | FencedRpcGateway.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/FencedRpcGateway.java` | フェンシング対応Gateway |

**読解のコツ**: RpcGateway はマーカーインターフェースとして機能し、実際のRPCメソッドは継承インターフェースで定義される。

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

処理の起点となる RpcService と RpcEndpoint を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | RpcService.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcService.java` | サービスインターフェース。connect()、startServer() |
| 2-2 | RpcEndpoint.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java` | エンドポイント基底クラス。ライフサイクル管理 |

**主要処理フロー**:
1. **142-153行目**: コンストラクタで RpcService.startServer() を呼び出し
2. **201-203行目**: start() でサーバーを開始状態に
3. **306-309行目**: closeAsync() で非同期終了

#### Step 3: Pekko実装を理解する

実際の通信処理を担う Pekko ベースの実装を読む。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PekkoRpcService.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcService.java` | RpcService の Pekko 実装 |
| 3-2 | PekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java` | RPC処理を行う Actor |
| 3-3 | PekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoInvocationHandler.java` | プロキシのInvocationHandler |

**主要処理フロー（PekkoRpcService）**:
- **262-337行目**: startServer() - Actor生成とプロキシ作成
- **214-234行目**: connect() - リモートエンドポイント接続
- **489-543行目**: connectInternal() - ハンドシェイクとプロキシ生成

**主要処理フロー（PekkoRpcActor）**:
- **160-166行目**: createReceive() - メッセージハンドラ定義
- **285-347行目**: handleRpcInvocation() - RPC呼び出し処理
- **453-483行目**: handleRunAsync() - 非同期Runnable実行

#### Step 4: メッセージ型を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | LocalRpcInvocation.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/messages/LocalRpcInvocation.java` | ローカル呼び出し用メッセージ |
| 4-2 | RemoteRpcInvocation.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/messages/RemoteRpcInvocation.java` | リモート呼び出し用（シリアライズ対応） |
| 4-3 | RunAsync.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/messages/RunAsync.java` | 非同期Runnable実行メッセージ |

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

```
RpcEndpoint (ユーザーコンポーネント)
    │
    ├─ RpcService.startServer()
    │      └─ PekkoRpcService.startServer()
    │             ├─ SupervisorActor.startRpcActor()
    │             │      └─ PekkoRpcActor 生成
    │             └─ Proxy.newProxyInstance()
    │                    └─ PekkoInvocationHandler
    │
    └─ RpcService.connect()
           └─ PekkoRpcService.connectInternal()
                  ├─ ActorSystem.actorSelection()
                  ├─ Patterns.ask() (ハンドシェイク)
                  └─ Proxy.newProxyInstance()
                         └─ PekkoInvocationHandler

RpcGateway (プロキシ呼び出し)
    │
    └─ PekkoInvocationHandler.invoke()
           ├─ invokeRpc() (RPC呼び出し)
           │      ├─ createRpcInvocationMessage()
           │      └─ ask() / tell()
           │             └─ PekkoRpcActor.handleRpcInvocation()
           │                    └─ Method.invoke()
           │
           └─ runAsync() / callAsync() (非同期実行)
                  └─ tell(RunAsync/CallAsync)
                         └─ PekkoRpcActor.handleRunAsync/handleCallAsync()
```

### データフロー図

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

RPC呼び出し           PekkoInvocationHandler              CompletableFuture
(メソッド呼び出し) ──▶ ├─ RpcInvocation生成 ────▶    (非同期結果)
                      └─ Actor.tell/ask()

                            │
                            ▼

                      PekkoRpcActor
                      ├─ メッセージ受信
                      ├─ メソッドルックアップ
                      └─ Method.invoke()
                            │
                            ▼

                      RpcEndpoint
                      (実際の処理実行)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| RpcService.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcService.java` | ソース | RPCサービスインターフェース |
| RpcEndpoint.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java` | ソース | エンドポイント基底クラス |
| RpcGateway.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java` | ソース | Gatewayインターフェース |
| RpcServer.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java` | ソース | サーバーインターフェース |
| FencedRpcEndpoint.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/FencedRpcEndpoint.java` | ソース | フェンシング対応エンドポイント |
| RpcInvocation.java | `flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/messages/RpcInvocation.java` | ソース | RPC呼び出しメッセージ |
| PekkoRpcService.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcService.java` | ソース | Pekko RpcService実装 |
| PekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java` | ソース | RPCメッセージ処理Actor |
| PekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoInvocationHandler.java` | ソース | プロキシInvocationHandler |
| SupervisorActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/SupervisorActor.java` | ソース | Actor監督者 |
| ControlMessages.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/ControlMessages.java` | ソース | 制御メッセージ定義 |
