# 機能設計書 72-Pekko RPC

## 概要

本ドキュメントは、Apache Flink の Pekko RPC 実装に関する機能設計書である。flink-rpc-akka モジュールで提供される Apache Pekko（旧 Akka）ベースの非同期RPC実装について、その構造、処理フロー、および実装詳細を記載する。

### 本機能の処理概要

Pekko RPC は、Flink の RpcService インターフェースの標準実装であり、Apache Pekko のアクターモデルを活用して、分散コンポーネント間の非同期通信を実現する。

**業務上の目的・背景**：Flink のRPCシステムは抽象化されたインターフェースとして設計されており、実際の通信処理は具体的な実装に委譲される。Pekko（Akka のフォーク）は、高性能な分散アクターシステムとして実績があり、メッセージパッシング、障害監視、位置透過性といった機能を提供する。Flink はこの Pekko を活用することで、効率的で信頼性の高いコンポーネント間通信を実現している。

**機能の利用シーン**：
- JobManager と TaskManager 間の全ての通信
- クラスターコーディネーション（リーダー選出、リソース割当）
- チェックポイント調整メッセージの送受信
- ハートビートによる生存監視

**主要な処理内容**：
1. ActorSystem の初期化と設定
2. RpcEndpoint 用 Actor（PekkoRpcActor）の生成と管理
3. 動的プロキシ（PekkoInvocationHandler）によるRPC呼び出しの透過化
4. メッセージのシリアライズとフレームサイズ制御
5. 接続確立時のハンドシェイクプロトコル

**関連システム・外部連携**：
- Apache Pekko アクターシステム
- Netty ネットワーク層（Pekko Remote）
- Java Serialization / Kryo シリアライザ

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

## 関連画面

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| actorSystem | ActorSystem | Yes | Pekko アクターシステム | null不可 |
| configuration | PekkoRpcServiceConfiguration | Yes | RPC設定（タイムアウト、フレームサイズ等） | null不可 |
| flinkClassLoader | ClassLoader | Yes | Flinkのクラスローダー | null不可 |
| rpcEndpoint | RpcEndpoint | Yes | RPCエンドポイント実装 | RpcGatewayも実装必須 |
| address | String | Yes | 接続先アドレス（pekko://...形式） | 有効なPekko URL |

### 入力データソース

- Flink コンポーネント（JobMaster、ResourceManager、TaskExecutor など）
- 設定ファイル（flink-conf.yaml）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| RpcServer | RpcServer | 起動されたRPCサーバープロキシ |
| RpcGateway | C extends RpcGateway | 接続されたリモートGatewayプロキシ |
| ActorRef | ActorRef | 生成されたアクター参照 |

### 出力先

- 呼び出し元コンポーネント

## 処理フロー

### 処理シーケンス

```
1. PekkoRpcService の初期化
   └─ ActorSystem から address/port 取得
   └─ SupervisorActor 起動
   └─ DeadLettersActor 起動

2. RpcEndpoint サーバー起動 (startServer)
   └─ registerRpcActor で PekkoRpcActor 生成
   └─ InvocationHandler 作成（PekkoInvocationHandler または FencedPekkoInvocationHandler）
   └─ Proxy.newProxyInstance でプロキシ生成

3. リモートエンドポイント接続 (connect)
   └─ ActorSelection で対象 Actor 解決
   └─ RemoteHandshakeMessage でハンドシェイク
   └─ HandshakeSuccessMessage 受信後プロキシ生成

4. RPC 呼び出し実行
   └─ PekkoInvocationHandler.invoke()
   └─ RpcInvocation メッセージ生成
   └─ Patterns.ask() で非同期送信
   └─ PekkoRpcActor で受信・処理
   └─ 結果を CompletableFuture で返却

5. サービス終了
   └─ 全 Actor の終了
   └─ ActorSystem.terminate()
```

### フローチャート

```mermaid
flowchart TD
    A[PekkoRpcService 生成] --> B[ActorSystem 初期化]
    B --> C[SupervisorActor 起動]
    C --> D[DeadLettersActor 起動]

    E[startServer 呼び出し] --> F{Fenced?}
    F -->|Yes| G[FencedPekkoRpcActor 生成]
    F -->|No| H[PekkoRpcActor 生成]
    G --> I[FencedPekkoInvocationHandler 作成]
    H --> J[PekkoInvocationHandler 作成]
    I --> K[Proxy.newProxyInstance]
    J --> K
    K --> L[RpcServer 返却]

    M[connect 呼び出し] --> N[actorSelection]
    N --> O[resolveOne]
    O --> P[ハンドシェイク送信]
    P --> Q{成功?}
    Q -->|Yes| R[Proxy 生成・返却]
    Q -->|No| S[RpcConnectionException]

    T[RPC メソッド呼び出し] --> U[invoke]
    U --> V[RpcInvocation 生成]
    V --> W{void 戻り値?}
    W -->|Yes| X[tell で送信]
    W -->|No| Y[ask で送信]
    Y --> Z[CompletableFuture 返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-72-01 | バージョン互換 | ハンドシェイク時にRPCバージョン（VERSION=2）を検証 | 接続時 |
| BR-72-02 | フレームサイズ制限 | レスポンスが maximumFramesize を超える場合はエラー | 常時 |
| BR-72-03 | シリアライズ強制 | forceRpcInvocationSerialization が true の場合、ローカルでもシリアライズ | 設定時 |
| BR-72-04 | @Local アノテーション | @Local 付きメソッドはローカル呼び出し扱い | メソッド定義時 |
| BR-72-05 | 状態遷移 | Actor は STOPPED -> STARTED -> TERMINATING -> TERMINATED の順で遷移 | 常時 |

### 計算ロジック

特になし

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RpcConnectionException | 接続エラー | Actor 解決失敗 | アドレス確認、ネットワーク確認 |
| HandshakeException | ハンドシェイクエラー | バージョン不一致、Gateway非対応 | Flinkバージョン統一 |
| RpcException | フレームサイズ超過 | レスポンスが制限を超える | データサイズ削減、設定変更 |
| AskTimeoutException | タイムアウト | 応答時間超過 | タイムアウト値調整 |
| UnknownMessageException | 不明メッセージ | 想定外のメッセージ受信 | コード確認 |
| EndpointNotStartedException | 未起動 | 起動前にメッセージ受信 | 起動順序確認 |

### リトライ仕様

Pekko レベルでの自動リトライは行わない。アプリケーション層で必要に応じて実装する。

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

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

## パフォーマンス要件

- メッセージ処理: マイクロ秒〜ミリ秒オーダー
- 同時接続数: ActorSystem のスレッドプール設定に依存
- フレームサイズ: デフォルト設定に従う

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

- CustomSSLEngineProvider による SSL/TLS 通信
- Pekko Remote のセキュリティ設定（設定ファイルで構成）

## 備考

- Pekko は Apache Akka 2.6 系のフォーク
- クラスローダー管理に注意（flinkClassLoader のコンテキスト設定）

---

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

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

### 推奨読解順序

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

まず、Pekko RPC で使用される設定と制御メッセージを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PekkoRpcServiceConfiguration.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcServiceConfiguration.java` | タイムアウト、フレームサイズ、シリアライズ強制などの設定 |
| 1-2 | ControlMessages.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/ControlMessages.java` | START, STOP, TERMINATE 制御メッセージ |
| 1-3 | RpcSerializedValue.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/RpcSerializedValue.java` | シリアライズされた値のラッパー |

**読解のコツ**: ControlMessages は enum で定義された制御メッセージ。Actor の状態遷移を制御する。

#### Step 2: サービス層を理解する

PekkoRpcService がどのようにサービスを提供するかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | PekkoRpcService.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcService.java` | RpcService 実装の中心 |

**主要処理フロー**:
1. **116-160行目**: コンストラクタ - ActorSystem 初期化、Supervisor/DeadLetters Actor 起動
2. **262-337行目**: startServer() - RpcEndpoint 用 Actor 生成とプロキシ作成
3. **214-234行目**: connect() - リモート接続（通常版）
4. **238-259行目**: connect() - リモート接続（Fenced版）
5. **489-543行目**: connectInternal() - 実際の接続処理

#### Step 3: Actor 層を理解する

PekkoRpcActor がどのようにメッセージを処理するかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java` | RPC メッセージ処理の中心 |
| 3-2 | FencedPekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/FencedPekkoRpcActor.java` | フェンシング対応版 |

**主要処理フロー（PekkoRpcActor）**:
- **160-166行目**: createReceive() - メッセージハンドラ定義
- **168-191行目**: handleMessage() - 一般メッセージ処理
- **193-212行目**: handleControlMessage() - 制御メッセージ処理
- **223-244行目**: handleRpcMessage() - RPC/RunAsync/CallAsync 振り分け
- **285-347行目**: handleRpcInvocation() - RPC 呼び出し実行
- **532-673行目**: State インターフェースと状態クラス群

#### Step 4: InvocationHandler を理解する

プロキシがどのように動作するかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | PekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoInvocationHandler.java` | 通常の InvocationHandler |
| 4-2 | FencedPekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/FencedPekkoInvocationHandler.java` | フェンシング対応版 |

**主要処理フロー（PekkoInvocationHandler）**:
- **125-149行目**: invoke() - プロキシメソッド呼び出しのエントリーポイント
- **217-286行目**: invokeRpc() - RPC 呼び出しの実装
- **299-317行目**: createRpcInvocationMessage() - メッセージ生成
- **340-344行目**: ask() - 非同期メッセージ送信

#### Step 5: 監督・ユーティリティを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | SupervisorActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/SupervisorActor.java` | Actor 監督・生成 |
| 5-2 | PekkoUtils.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java` | ユーティリティ関数群 |
| 5-3 | ActorSystemBootstrapTools.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/ActorSystemBootstrapTools.java` | ActorSystem 起動ユーティリティ |

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

```
PekkoRpcService
    │
    ├─ startServer(rpcEndpoint)
    │      │
    │      ├─ registerRpcActor()
    │      │      │
    │      │      └─ SupervisorActor.startRpcActor()
    │      │             └─ Props.create(PekkoRpcActor/FencedPekkoRpcActor)
    │      │
    │      └─ Proxy.newProxyInstance()
    │             └─ PekkoInvocationHandler / FencedPekkoInvocationHandler
    │
    ├─ connect(address, clazz)
    │      │
    │      └─ connectInternal()
    │             ├─ resolveActorAddress()
    │             │      └─ ActorSelection.resolveOne()
    │             │
    │             ├─ Patterns.ask(RemoteHandshakeMessage)
    │             │      └─ PekkoRpcActor.handleHandshakeMessage()
    │             │
    │             └─ Proxy.newProxyInstance()
    │
    └─ closeAsync()
           ├─ terminateRpcActors()
           │      └─ tell(TERMINATE)
           │
           └─ actorSystem.terminate()

PekkoRpcActor
    │
    ├─ createReceive()
    │      ├─ RemoteHandshakeMessage → handleHandshakeMessage()
    │      ├─ ControlMessages → handleControlMessage()
    │      └─ その他 → handleMessage()
    │
    ├─ handleMessage()
    │      └─ handleRpcMessage()
    │             ├─ RunAsync → handleRunAsync()
    │             ├─ CallAsync → handleCallAsync()
    │             └─ RpcInvocation → handleRpcInvocation()
    │
    └─ handleRpcInvocation()
           ├─ lookupRpcMethod()
           ├─ Method.invoke()
           └─ sendSyncResponse() / sendAsyncResponse()
```

### データフロー図

```
[クライアント側]              [通信層]                    [サーバー側]

RpcGateway                                            PekkoRpcActor
(Proxy)                                               (AbstractActor)
    │                                                      │
    │   invoke()                                           │
    ▼                                                      │
PekkoInvocationHandler                                     │
    │                                                      │
    ├─ createRpcInvocationMessage()                        │
    │      │                                               │
    │      ▼                                               │
    │   LocalRpcInvocation                                 │
    │   / RemoteRpcInvocation                              │
    │                                                      │
    └─ ask(rpcInvocation)                                  │
           │                                               │
           │     [Pekko Message Passing]                   │
           └─────────────────────────────────────────────▶ │
                                                           │
                                           handleRpcInvocation()
                                                  │
                                                  ├─ lookupRpcMethod()
                                                  ├─ Method.invoke()
                                                  │      │
                                                  │      ▼
                                                  │   RpcEndpoint
                                                  │   (実処理)
                                                  │
                                           ◀──────┘
                                           │
                                    sendSyncResponse()
                                    / sendAsyncResponse()
                                           │
    ◀──────────────────────────────────────┘
    │
CompletableFuture
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| PekkoRpcService.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcService.java` | ソース | RpcService の Pekko 実装 |
| PekkoRpcServiceConfiguration.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcServiceConfiguration.java` | ソース | 設定クラス |
| PekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java` | ソース | RPC メッセージ処理 Actor |
| FencedPekkoRpcActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/FencedPekkoRpcActor.java` | ソース | フェンシング対応 Actor |
| PekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoInvocationHandler.java` | ソース | プロキシ InvocationHandler |
| FencedPekkoInvocationHandler.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/FencedPekkoInvocationHandler.java` | ソース | フェンシング対応 Handler |
| 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` | ソース | 制御メッセージ enum |
| RpcSerializedValue.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/RpcSerializedValue.java` | ソース | シリアライズ値ラッパー |
| PekkoUtils.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoUtils.java` | ソース | ユーティリティ |
| ActorSystemBootstrapTools.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/ActorSystemBootstrapTools.java` | ソース | ActorSystem 起動ツール |
| DeadLettersActor.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/DeadLettersActor.java` | ソース | 配達不能メッセージ処理 |
| PekkoBasedEndpoint.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoBasedEndpoint.java` | ソース | ActorRef アクセスインターフェース |
| CustomSSLEngineProvider.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/CustomSSLEngineProvider.java` | ソース | SSL エンジン提供 |
| RobustActorSystem.java | `flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/RobustActorSystem.java` | ソース | 堅牢な ActorSystem 実装 |
