# 機能設計書 11-非同期I/O

## 概要

本ドキュメントは、Apache Flink の非同期I/O（AsyncDataStream）機能について、その設計と実装の詳細を記述したものである。非同期I/Oは、ストリーム処理において外部システムへの非同期アクセスを実現するための機能である。

### 本機能の処理概要

非同期I/O機能は、ストリーム処理中に外部システム（データベース、REST API、キャッシュサービス等）への問い合わせを非同期に実行することで、ストリーム処理のスループットを大幅に向上させる機能である。

**業務上の目的・背景**：
従来の同期的な外部システムアクセスでは、各レコードの処理が外部システムからの応答を待つ間ブロックされ、全体のスループットが大幅に低下する。特に、ネットワークレイテンシが高い場合やバッチ処理が必要な外部システムへのアクセスにおいて、この問題は顕著である。非同期I/O機能は、複数の外部システムリクエストを並行して発行し、応答を非同期に処理することで、この問題を解決する。

**機能の利用シーン**：
- リアルタイムストリーム処理でのデータベース参照（エンリッチメント処理）
- 外部REST APIへの問い合わせを含むストリーム処理
- Redis/Memcached等のキャッシュシステムへのルックアップ
- 機械学習モデルサービスへの推論リクエスト

**主要な処理内容**：
1. AsyncDataStream.orderedWait/unorderedWait メソッドによるオペレータの追加
2. AsyncFunction.asyncInvoke メソッドでの非同期処理の実行
3. ResultFuture を通じた非同期結果の収集
4. タイムアウト管理とリトライ機構
5. チェックポイント時のステート保存と復旧

**関連システム・外部連携**：
外部データベース、REST API、キャッシュサービス、メッセージキュー等、あらゆる非同期アクセス可能な外部システムとの連携が可能である。

**権限による制御**：
特別な権限制御はなく、Flinkジョブを実行できるユーザーであれば利用可能である。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面との直接的な関連はない（プログラムAPIとしての機能） |

## 機能種別

データ連携 / ストリーム変換処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| in | DataStream<IN> | Yes | 入力データストリーム | null不可 |
| func | AsyncFunction<IN, OUT> | Yes | 非同期処理を行うユーザー定義関数 | null不可 |
| timeout | long | Yes | 非同期操作のタイムアウト時間 | リトライ有効時は0より大きい値 |
| timeUnit | TimeUnit | Yes | タイムアウトの時間単位 | null不可 |
| capacity | int | No | バッファ容量（デフォルト: 100） | 0より大きい値 |
| asyncRetryStrategy | AsyncRetryStrategy<OUT> | No | リトライ戦略 | - |

### 入力データソース

DataStream APIを通じて生成されたデータストリーム

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| result | SingleOutputStreamOperator<OUT> | 非同期処理結果を含む出力ストリーム |

### 出力先

後続のDataStream変換処理またはSink

## 処理フロー

### 処理シーケンス

```
1. AsyncDataStream.orderedWait/unorderedWaitの呼び出し
   └─ AsyncWaitOperatorFactoryを生成し、ストリームに変換を追加

2. レコード受信時（processElement）
   └─ キューにエントリを追加
   └─ AsyncFunction.asyncInvokeを呼び出し
   └─ タイムアウトタイマーを登録

3. 非同期処理完了時
   └─ ResultFuture.completeにより結果を通知
   └─ 完了したエレメントを出力キューへ移動

4. 出力処理
   └─ 完了エレメントをタイムスタンプ付きコレクターへ出力
   └─ MailboxExecutorを通じて順次出力

5. チェックポイント処理
   └─ インフライトエレメントをステートに保存
   └─ 復旧時にステートからリプレイ
```

### フローチャート

```mermaid
flowchart TD
    A[レコード受信] --> B{キューに空きあり?}
    B -->|No| C[Mailboxでyield]
    C --> B
    B -->|Yes| D[キューにエントリ追加]
    D --> E[asyncInvoke呼び出し]
    E --> F{リトライ有効?}
    F -->|Yes| G[RetryableResultHandlerDelegator作成]
    F -->|No| H[ResultHandler作成]
    G --> I[タイムアウトタイマー登録]
    H --> I
    I --> J[非同期処理実行]
    J --> K{処理成功?}
    K -->|Yes| L[ResultFuture.complete]
    K -->|No| M{リトライ条件?}
    M -->|Yes| N[リトライスケジュール]
    M -->|No| O[ResultFuture.completeExceptionally]
    N --> J
    L --> P[完了エレメント出力]
    O --> Q[エラー処理]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 出力順序保証（Ordered） | 入力順序と同じ順序で出力する | OutputMode.ORDEREDの場合 |
| BR-02 | 出力順序非保証（Unordered） | 処理完了順に出力する（スループット優先） | OutputMode.UNORDEREDの場合 |
| BR-03 | タイムアウト処理 | タイムアウト発生時はAsyncFunction.timeoutを呼び出す | timeout > 0 の場合 |
| BR-04 | リトライ条件 | 結果または例外がリトライ条件に合致する場合にリトライ | AsyncRetryStrategyが設定されている場合 |

### 計算ロジック

リトライ判定ロジック:
- 結果がretryResultPredicateに合致する場合、リトライ対象
- 例外がretryExceptionPredicateに合致する場合、リトライ対象
- currentAttempts < maxAttempts かつ タイムアウト未到達の場合、リトライ実行

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

本機能は直接的なデータベース操作を行わない。ステート管理はFlinkのオペレーターステートを使用する。

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| チェックポイント | オペレーターステート | WRITE | インフライトエレメントの保存 |
| リストア | オペレーターステート | READ | 保存されたエレメントの復元 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TimeoutException | タイムアウト | 非同期処理がタイムアウト時間を超過 | AsyncFunction.timeoutメソッドで処理 |
| Exception | ユーザーエラー | asyncInvoke内でのユーザーコードエラー | タスクのフェイルオーバー |
| RejectedExecutionException | システムエラー | オペレーターキャンセル時 | 処理をスキップ |

### リトライ仕様

AsyncRetryStrategyインターフェースにより設定可能:
- 最大リトライ回数
- リトライ間隔（バックオフ）
- リトライ条件（結果ベース/例外ベース）

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

本機能自体はトランザクション管理を行わない。チェックポイントとの統合により、Exactly-once処理を保証する場合は、外部システム側での冪等性確保が必要である。

## パフォーマンス要件

- デフォルトキュー容量: 100（同時に処理可能な非同期リクエスト数）
- キュー容量はメモリ使用量とスループットのトレードオフ
- Ordered/Unorderedモードにより出力レイテンシが変化

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

- 外部システムへの認証情報はAsyncFunction内で安全に管理する必要がある
- シリアライズ可能なAsyncFunctionはステートに含まれる可能性があるため、機密情報の取り扱いに注意

## 備考

- Watermarkはキューを通過する際に完了マークが付与される
- 入力終了時（endInput）には、全インフライトリトライを即座に完了させる

---

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

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

### 推奨読解順序

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

まず、非同期処理で受け渡されるデータ構造と結果通知インターフェースを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ResultFuture.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java` | 非同期結果を通知するためのインターフェース |
| 1-2 | StreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueue.java` | 非同期処理キューのインターフェース |
| 1-3 | AsyncRetryStrategy.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncRetryStrategy.java` | リトライ戦略の定義 |

**読解のコツ**: ResultFutureはJava 8のCompletableFutureに似た設計だが、Flink独自のシンプルなインターフェースとして定義されている。

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

ユーザーが非同期I/Oを使用する際のAPIを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AsyncDataStream.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AsyncDataStream.java` | 非同期I/Oの静的APIエントリーポイント |
| 2-2 | AsyncFunction.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java` | ユーザー実装インターフェース |

**主要処理フロー**:
1. **68-101行目（AsyncDataStream.java）**: addOperatorメソッドでAsyncWaitOperatorFactoryを生成
2. **115-128行目**: unorderedWaitメソッドでUNORDEREDモードの非同期オペレータを追加
3. **165-178行目**: orderedWaitメソッドでORDEREDモードの非同期オペレータを追加
4. **87行目（AsyncFunction.java）**: asyncInvokeメソッドがユーザー実装の中核

#### Step 3: オペレータ実装を理解する

非同期処理の中核となるオペレータの実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AsyncWaitOperator.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java` | 非同期I/Oのメインオペレータ |
| 3-2 | OrderedStreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/OrderedStreamElementQueue.java` | 順序保証キュー |
| 3-3 | UnorderedStreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/UnorderedStreamElementQueue.java` | 順序非保証キュー |

**主要処理フロー**:
- **245-278行目（AsyncWaitOperator.java）**: processElementメソッドでレコードを処理キューに追加し非同期呼び出しを実行
- **291-311行目**: snapshotStateメソッドでチェックポイント時にキュー内容を保存
- **391-411行目**: outputCompletedElementメソッドで完了したエレメントを出力
- **435-591行目**: RetryableResultHandlerDelegatorでリトライロジックを実装

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

```
AsyncDataStream (API)
    │
    ├─ addOperator()
    │      └─ AsyncWaitOperatorFactory (ファクトリ)
    │             └─ AsyncWaitOperator (オペレータ)
    │
    └─ AsyncWaitOperator
           │
           ├─ processElement()
           │      ├─ addToWorkQueue() → StreamElementQueue
           │      └─ AsyncFunction.asyncInvoke()
           │             └─ ResultHandler / RetryableResultHandlerDelegator
           │                    └─ ResultFuture.complete()
           │
           ├─ snapshotState()
           │      └─ ListState保存
           │
           └─ outputCompletedElement()
                  └─ TimestampedCollector出力
```

### データフロー図

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

DataStream<IN>  ───▶  AsyncWaitOperator  ───▶  SingleOutputStreamOperator<OUT>
                           │
                           │ asyncInvoke()
                           ▼
                    ┌─────────────────┐
                    │  External       │
                    │  System         │
                    │ (DB/API/Cache)  │
                    └─────────────────┘
                           │
                           │ ResultFuture.complete()
                           ▼
                    StreamElementQueue
                    (Ordered/Unordered)
                           │
                           ▼
                    TimestampedCollector
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AsyncDataStream.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AsyncDataStream.java` | ソース | 非同期I/OのAPI |
| AsyncFunction.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java` | ソース | ユーザー実装インターフェース |
| RichAsyncFunction.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/RichAsyncFunction.java` | ソース | ライフサイクル管理付きAsyncFunction |
| ResultFuture.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java` | ソース | 結果通知インターフェース |
| AsyncRetryStrategy.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncRetryStrategy.java` | ソース | リトライ戦略 |
| AsyncRetryPredicate.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncRetryPredicate.java` | ソース | リトライ条件 |
| AsyncWaitOperator.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java` | ソース | メインオペレータ |
| AsyncWaitOperatorFactory.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorFactory.java` | ソース | オペレータファクトリ |
| StreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueue.java` | ソース | キューインターフェース |
| OrderedStreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/OrderedStreamElementQueue.java` | ソース | 順序保証キュー実装 |
| UnorderedStreamElementQueue.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/UnorderedStreamElementQueue.java` | ソース | 非順序キュー実装 |
| StreamRecordQueueEntry.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamRecordQueueEntry.java` | ソース | キューエントリ |
| WatermarkQueueEntry.java | `flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/WatermarkQueueEntry.java` | ソース | Watermark用エントリ |
