# 通知設計書 33-notifyReadyAsync

## 概要

本ドキュメントは、Apache Flinkにおける非同期準備完了通知（notifyReadyAsync）の設計仕様を記述したものである。

### 本通知の処理概要

notifyReadyAsyncは、ExecutorNotifierクラスで提供される非同期通知メカニズムである。このメソッドは、ワーカースレッドで実行される処理が完了した後、その結果をメインスレッド（またはコーディネータスレッド）に通知するためのコールシーションパターンを実装している。主にSourceCoordinatorContextで使用され、SplitEnumeratorの処理結果をコーディネータスレッドに安全に伝達する。

**業務上の目的・背景**：Flinkのソースコネクタアーキテクチャでは、SplitEnumeratorがソースデータの分割を管理し、各ソースリーダーにスプリットを割り当てる。この処理は複雑でブロッキングになる可能性があるため、ワーカースレッドで実行される。しかし、処理結果（スプリットの割り当て情報など）はメインのコーディネータスレッドで処理される必要がある。notifyReadyAsyncは、この異なるスレッド間のデータ伝達を安全かつ効率的に行うための仕組みを提供する。

**通知の送信タイミング**：notifyReadyAsyncメソッドが呼び出されると、ワーカースレッドプールで callable が実行され、その完了後に executorToNotify（通常はコーディネータエグゼキュータ）にハンドラの実行がスケジュールされる。オーバーロード版では、初期遅延と周期的実行も設定可能である。

**通知の受信者**：ExecutorNotifierのコンストラクタで指定されたexecutorToNotify（Executor）が通知の受信者となる。SourceCoordinatorContextでは、エラーハンドリング付きのcoordinatorExecutorがこの役割を担う。

**通知内容の概要**：callableの実行結果（型T）とエラー情報（Throwable）がBiConsumerハンドラに渡される。正常完了時はresultにcallableの戻り値が、異常時はthrowableにエラー情報が設定される。

**期待されるアクション**：ハンドラは通知を受けて、callableの結果に基づいた処理を実行する。例えば、スプリット割り当て結果をソースリーダーに送信したり、エラー発生時には適切なエラーハンドリングを行う。重要な点として、状態の変更はすべてハンドラ内で行い、callable内では行わないことが推奨される（スレッドセーフティのため）。

## 通知種別

内部コールバック通知（Executor経由の非同期コールバック）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（ワーカースレッドからコーディネータスレッドへ） |
| 優先度 | 中 |
| リトライ | なし（エラーはハンドラに伝達） |

### 送信先決定ロジック

ExecutorNotifierのコンストラクタで指定されたexecutorToNotifyに通知される。SourceCoordinatorContextでは、ThrowableCatchingRunnableでラップされたerrorHandlingCoordinatorExecutorが使用され、未捕捉例外が発生した場合はhandleUncaughtExceptionFromAsyncCallで処理される。

## 通知テンプレート

### コールバック通知の場合

| 項目 | 内容 |
|-----|------|
| クラス | ExecutorNotifier |
| メソッドシグネチャ | void notifyReadyAsync(Callable<T> callable, BiConsumer<T, Throwable> handler) |
| オーバーロード | void notifyReadyAsync(Callable<T> callable, BiConsumer<T, Throwable> handler, long initialDelayMs, long periodMs) |

### メソッド定義

```java
/**
 * @param callable ワーカースレッドで実行される処理
 * @param handler 結果を受け取るハンドラ（コーディネータスレッドで実行）
 */
public <T> void notifyReadyAsync(Callable<T> callable, BiConsumer<T, Throwable> handler);

/**
 * @param callable ワーカースレッドで実行される処理
 * @param handler 結果を受け取るハンドラ
 * @param initialDelayMs 初期遅延（ミリ秒）
 * @param periodMs 実行周期（ミリ秒）
 */
public <T> void notifyReadyAsync(
    Callable<T> callable, BiConsumer<T, Throwable> handler,
    long initialDelayMs, long periodMs);
```

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| callable | ワーカースレッドで実行する処理 | 呼び出し元が提供 | Yes |
| handler | 結果を処理するBiConsumer | 呼び出し元が提供 | Yes |
| initialDelayMs | 初期遅延（周期実行版のみ） | 呼び出し元が提供 | No |
| periodMs | 実行周期（周期実行版のみ） | 呼び出し元が提供 | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 即時実行 | notifyReadyAsync(callable, handler) | なし | callableをワーカースレッドで即座に実行 |
| 周期実行 | notifyReadyAsync(callable, handler, delay, period) | なし | 初期遅延後、周期的にcallableを実行 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| Executorシャットダウン | workerExecutorまたはexecutorToNotifyがシャットダウン済みの場合、例外がスローされる可能性がある |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[notifyReadyAsync呼び出し] --> B[workerExecutor.execute]
    B --> C[ワーカースレッドでcallable.call]
    C --> D{例外発生?}
    D -->|No| E[result = callable戻り値]
    E --> F[executorToNotify.execute]
    F --> G[handler.accept result, null]
    D -->|Yes| H[throwable = 例外]
    H --> I[executorToNotify.execute]
    I --> J[handler.accept null, throwable]
    G --> K[終了]
    J --> K
```

### 周期実行版のフロー

```mermaid
flowchart TD
    A[notifyReadyAsync with period呼び出し] --> B[workerExecutor.scheduleAtFixedRate]
    B --> C[initialDelayMs待機]
    C --> D[callable.call実行]
    D --> E[executorToNotify.execute handler]
    E --> F[periodMs待機]
    F --> D
```

## データベース参照・更新仕様

### 参照テーブル一覧

本通知はインメモリの非同期通信パターンで実装されており、直接的なデータベースアクセスは行わない。

| 対象 | 用途 | 備考 |
|------|------|------|
| なし | - | - |

### 更新テーブル一覧

| 対象 | 操作 | 概要 |
|------|------|------|
| なし | - | - |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| callable例外 | callable.call()が例外をスローした場合 | handler.accept(null, throwable)で伝達 |
| handler例外 | ハンドラ内で例外が発生した場合 | ThrowableCatchingRunnableで捕捉（SourceCoordinatorContext使用時） |
| RejectedExecutionException | Executorがシャットダウン済みの場合 | 呼び出し元に例外がスロー |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（周期実行版は継続） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし

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

- 通知はJVMプロセス内の内部通信であり、ネットワーク経由では送信されない
- callableとhandlerで共有状態を変更する場合はスレッドセーフティに注意が必要
- Javadocで明示されているように、状態変更はハンドラ内で行うことが推奨される

## 備考

- 同じnotifyReadyAsyncが複数回呼び出された場合、複数のcallableが並行実行される可能性がある
- handlerも複数が並行実行される可能性があるため、ConcurrentModificationExceptionに注意
- 推奨パターン：callableは計算のみ、状態変更はハンドラで行う
- SourceCoordinatorContextでは、coordinatorExecutorとworkerExecutorの2つのスレッドプールが使用される

---

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

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

### 推奨読解順序

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

まず、ExecutorNotifierクラスの構造を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ExecutorNotifier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/ExecutorNotifier.java` | ワーカースレッドとコーディネータスレッド間の通知メカニズム |

**読解のコツ**: Javadocのサンプルコードが重要。callableとhandlerで状態を共有する際のアンチパターンと推奨パターンが示されている。

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

ExecutorNotifierの実装を詳しく確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ExecutorNotifier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/ExecutorNotifier.java` | notifyReadyAsyncメソッドの実装 |

**主要処理フロー**:
1. **34-42行目**: コンストラクタでworkerExecutorとexecutorToNotifyを受け取る
2. **44-75行目**: notifyReadyAsync（即時実行版）のJavadocとサンプルコード
3. **76-86行目**: notifyReadyAsync（即時実行版）の実装
   - **77-78行目**: workerExecutor.executeでcallableを実行
   - **80-81行目**: 正常終了時にexecutorToNotify.execute(handler.accept(result, null))
   - **82-83行目**: 例外時にexecutorToNotify.execute(handler.accept(null, t))
4. **88-121行目**: notifyReadyAsync（周期実行版）のJavadocと実装
   - **127-138行目**: scheduleAtFixedRateで周期実行

#### Step 3: 利用側のコンテキストを理解する

SourceCoordinatorContextでの利用を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SourceCoordinatorContext.java | `flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java` | ExecutorNotifierの生成と利用 |

**主要処理フロー**:
- **108-110行目**: workerExecutorとcoordinatorExecutorの定義
- **166-172行目**: errorHandlingCoordinatorExecutorの生成（ThrowableCatchingRunnableでラップ）
- **172行目**: ExecutorNotifierのインスタンス化

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

```
[呼び出し元（例：SplitEnumerator）]
        │
        ▼
ExecutorNotifier.notifyReadyAsync(callable, handler)
        │
        └─ workerExecutor.execute(...)
               │
               ▼
        [ワーカースレッド]
               │
               ├─ T result = callable.call()
               │
               └─ executorToNotify.execute(...)
                      │
                      ▼
               [コーディネータスレッド]
                      │
                      └─ handler.accept(result, null)
                             │
                             └─ [結果に基づく処理]
```

### データフロー図

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

callable ─────────▶ workerExecutor ─────────▶ result
                         │
                         ▼
                   executorToNotify
                         │
                         ▼
                   handler.accept(result, throwable)
                         │
                         ▼
                   [状態更新・イベント送信等]
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ExecutorNotifier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/ExecutorNotifier.java` | クラス | ワーカー/コーディネータ間の非同期通知メカニズム |
| SourceCoordinatorContext.java | `flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java` | クラス | SplitEnumeratorのコンテキスト、ExecutorNotifierを利用 |
| ExecutorNotifierTest.java | `flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/ExecutorNotifierTest.java` | テスト | ExecutorNotifierのユニットテスト |
| ThrowableCatchingRunnable.java | `flink-util/src/main/java/org/apache/flink/util/ThrowableCatchingRunnable.java` | ユーティリティ | 例外捕捉ラッパー |
