# 通知設計書 35-notifyDataAvailable（GateNotificationHelper）

## 概要

本ドキュメントは、Apache Flinkにおける入力ゲートデータ利用可能通知（notifyDataAvailable）の設計仕様を記述したものである。

### 本通知の処理概要

notifyDataAvailableは、GateNotificationHelperクラスで提供されるデータ利用可能通知メソッドである。InputGateにおいて新しいデータ（バッファ）が利用可能になった際に、待機中のタスクスレッドに通知する役割を担う。この通知により、タスクはブロッキング待機から復帰してデータを処理できる。

**業務上の目的・背景**：Flinkのストリーム処理では、タスクは入力データの到着を待機しながら処理を行う。InputGateは上流タスクからのデータを受信するためのインターフェースであり、データが利用可能になったタイミングでタスクに通知する必要がある。notifyDataAvailableは、この通知を効率的に行うための仕組みを提供し、タスクが無駄なポーリングを行わずにデータ到着を検知できるようにする。

**通知の送信タイミング**：SingleInputGateまたはUnionInputGateのqueueChannel/queueInputGateメソッド内で、データを持つチャネル/ゲートがキューに追加され、かつそれが最初の要素である場合に通知される。具体的には、inputChannelsWithData.size() == 1 の条件を満たす時に呼び出される。

**通知の受信者**：InputGateのavailabilityHelper.getUnavailableToResetAvailable()で取得されるCompletableFutureを待機しているスレッドが受信者となる。また、availabilityMonitor.notifyAll()により、Object.wait()で待機しているスレッドにも通知される。

**通知内容の概要**：notifyDataAvailable()メソッドは、2つの方法で通知を行う：(1) availabilityMonitor.notifyAll()による同期的なスレッド通知、(2) availabilityHelperからCompletableFutureを取得しcomplete(null)で完了させる非同期通知。通知自体にはデータは含まれず、「データが利用可能になった」という事実のみを伝達する。

**期待されるアクション**：受信者（タスクスレッド）はFutureの完了またはnotifyAll()を検知し、getNext()またはpollNext()でInputGateからデータを取得する。取得したデータはタスクの処理ロジックに渡され、ストリーム処理が継続される。

## 通知種別

内部コールバック通知（CompletableFuture + Object.notifyAll ベースの非同期通知）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（CompletableFuture.complete + Object.notifyAll） |
| 優先度 | 中 |
| リトライ | なし |

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

InputGateのavailabilityHelperが管理するCompletableFutureの待機者、およびavailabilityMonitorでObject.wait()している待機者に通知される。GateNotificationHelperはclose()メソッドでFutureをcomplete(null)することで通知を実行する。

## 通知テンプレート

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

| 項目 | 内容 |
|-----|------|
| クラス | GateNotificationHelper |
| メソッドシグネチャ | public void notifyDataAvailable() |

### メソッド定義

```java
/**
 * Must be called under lock to ensure integrity of availabilityHelper and allow notification.
 */
public void notifyDataAvailable() {
    availabilityMonitor.notifyAll();
    toNotify = inputGate.availabilityHelper.getUnavailableToResetAvailable();
}
```

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| availabilityMonitor | 同期待機用のモニターオブジェクト | コンストラクタで指定 | Yes |
| toNotify | 通知対象のCompletableFuture | inputGate.availabilityHelper | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| チャネルキュー | SingleInputGate.queueChannel | inputChannelsWithData.size() == 1 | 最初のデータチャネルがキューに追加された時 |
| ゲートキュー | UnionInputGate.queueInputGate | inputGatesWithData.size() == 1 | 最初のデータゲートがキューに追加された時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| データ要素が既に存在 | inputChannelsWithData.size() > 1 の場合 |
| EndOfPartitionイベント受信済み | チャネルがEndOfPartitionを受信している場合（SingleInputGate） |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[チャネル/ゲートにデータ到着] --> B[queueChannel/queueInputGate]
    B --> C[GateNotificationHelper作成]
    C --> D[synchronized inputChannelsWithData]
    D --> E[queueChannelUnsafe/キュー追加]
    E --> F{優先度イベント?}
    F -->|Yes| G[notifyPriority処理]
    F -->|No| H{最初の要素?}
    G --> H
    H -->|No| I[notifyDataAvailable呼び出しスキップ]
    H -->|Yes| J[notification.notifyDataAvailable]
    J --> K[availabilityMonitor.notifyAll]
    K --> L[toNotify = availabilityHelper.getUnavailableToResetAvailable]
    I --> M[ロック解放]
    L --> M
    M --> N[GateNotificationHelper.close]
    N --> O{toNotify != null?}
    O -->|Yes| P[toNotify.complete null]
    O -->|No| Q[終了]
    P --> Q
```

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

### 参照テーブル一覧

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

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| なし | 通常のCompletableFuture完了操作とnotifyAll()のため例外は発生しない | - |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（データ到着時に即座に通知）

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

- 通知はJVMプロセス内の内部通信であり、ネットワーク経由では送信されない
- notifyDataAvailable()はロック（inputChannelsWithData/inputGatesWithData）内で呼び出す必要がある
- 実際のFuture完了（close()）はロック外で行われ、デッドロックを防止

## 備考

- GateNotificationHelperはAutoCloseableを実装し、try-with-resourcesパターンで使用される
- notifyDataAvailableはnotifyPriorityと同様に2段階設計（ロック内準備、ロック外通知）
- availabilityMonitor.notifyAll()はロック内で呼び出されるが、これはInputGateの設計上のもの
- 同期待機（Object.wait）と非同期待機（CompletableFuture）の両方をサポート
- SingleInputGateではavailabilityMonitorとしてinputChannelsWithDataが使用される

---

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

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

### 推奨読解順序

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | GateNotificationHelper.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/GateNotificationHelper.java` | InputGateの利用可能性通知を抽象化するヘルパークラス |
| 1-2 | InputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java` | availabilityHelperの定義とgetAvailableFuture() |

**読解のコツ**: notifyDataAvailableはnotifyPriorityと同じGateNotificationHelperクラス内にあるが、availabilityMonitor.notifyAll()を追加で呼び出す点が異なる。

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

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | GateNotificationHelper.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/GateNotificationHelper.java` | notifyDataAvailableとclose()の実装 |

**主要処理フロー**:
1. **27-37行目**: フィールド定義とコンストラクタ
   - **28-29行目**: inputGateとavailabilityMonitorの保持
   - **31-32行目**: toNotifyPriorityとtoNotifyのCompletableFuture
2. **39-47行目**: close()メソッド
   - **44-46行目**: toNotifyがnullでなければcomplete(null)
3. **54-60行目**: notifyDataAvailable()メソッド
   - **58行目**: availabilityMonitor.notifyAll()の呼び出し
   - **59行目**: availabilityHelper.getUnavailableToResetAvailable()の呼び出し

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

SingleInputGateでの利用を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SingleInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java` | queueChannelでのGateNotificationHelper使用 |
| 3-2 | UnionInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java` | queueInputGateでのGateNotificationHelper使用 |

**主要処理フロー（SingleInputGate）**:
- **1209-1239行目**: queueChannelメソッド
  - **1211-1212行目**: try-with-resourcesでGateNotificationHelper作成
  - **1213行目**: synchronized (inputChannelsWithData)
  - **1234-1236行目**: size() == 1 の場合にnotification.notifyDataAvailable()

**主要処理フロー（UnionInputGate）**:
- **368-398行目**: queueInputGateメソッド
  - **371-372行目**: try-with-resourcesでGateNotificationHelper作成
  - **373行目**: synchronized (inputGatesWithData)
  - **393-395行目**: size() == 1 の場合にnotification.notifyDataAvailable()

#### Step 4: 待機側の処理を理解する

InputGateのgetNext()で待機がどのように解除されるかを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SingleInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java` | getChannel()メソッドでのwait処理 |

**主要処理フロー**:
- **1278-1298行目**: getChannel(boolean blocking)メソッド
  - **1281-1292行目**: inputChannelsWithData.isEmpty()の間ブロッキング待機
  - **1287行目**: inputChannelsWithData.wait()でnotifyAll()を待機

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

```
[ネットワークスレッド等]
        │
        ▼
SingleInputGate.queueChannel(channel, prioritySequenceNumber, forcePriority)
        │
        └─ try (GateNotificationHelper notification = new GateNotificationHelper(...))
               │
               ├─ synchronized (inputChannelsWithData)
               │      │
               │      ├─ queueChannelUnsafe(channel, priority)
               │      │
               │      ├─ [優先度イベントの場合] notification.notifyPriority()
               │      │
               │      └─ if (inputChannelsWithData.size() == 1)
               │             │
               │             └─ notification.notifyDataAvailable()
               │                    │
               │                    ├─ availabilityMonitor.notifyAll()
               │                    │      │
               │                    │      └─ [同期待機スレッドが起床]
               │                    │
               │                    └─ toNotify = availabilityHelper
               │                                  .getUnavailableToResetAvailable()
               │
               └─ [ロック解放後] notification.close()
                      │
                      └─ toNotify.complete(null)
                             │
                             └─ [非同期待機のFuture完了]
```

### データフロー図

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

データバッファ到着 ───▶ queueChannel/queueInputGate ───▶ inputChannelsWithData更新
                               │
                               ▼
                        GateNotificationHelper.notifyDataAvailable()
                               │
                               ├───▶ availabilityMonitor.notifyAll()
                               │          │
                               │          └───▶ Object.wait()からの復帰
                               │
                               └───▶ availabilityHelper
                                     .getUnavailableToResetAvailable()
                                          │
                                          ▼
                                     [ロック解放後]
                                     toNotify.complete(null)
                                          │
                                          └───▶ CompletableFuture待機からの復帰
                                                    │
                                                    ▼
                                          InputGate.getNext()でデータ取得
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| GateNotificationHelper.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/GateNotificationHelper.java` | クラス | InputGateの利用可能性通知ヘルパー |
| InputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java` | 抽象クラス | InputGateの基底クラス、availabilityHelper定義 |
| SingleInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java` | クラス | 単一の中間結果を消費するInputGate |
| UnionInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java` | クラス | 複数のInputGateを統合するInputGate |
| AvailabilityHelper.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/AvailabilityProvider.java` | ヘルパークラス | CompletableFutureベースの利用可能性管理 |
| PrioritizedDeque.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PrioritizedDeque.java` | クラス | 優先度付きキュー |
| BufferOrEvent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferOrEvent.java` | クラス | InputGateから取得されるデータ構造 |
