# 通知設計書 34-notifyPriority（GateNotificationHelper）

## 概要

本ドキュメントは、Apache Flinkにおける入力ゲート優先度通知（notifyPriority）の設計仕様を記述したものである。

### 本通知の処理概要

notifyPriorityは、GateNotificationHelperクラスで提供される優先度イベント通知メソッドである。InputGateにおいて優先度の高いイベント（チェックポイントバリアなど）が利用可能になった際に、待機中のタスクスレッドに通知する役割を担う。この通知により、タスクは通常のデータ処理を中断して優先度イベントを先に処理できる。

**業務上の目的・背景**：Flinkのチェックポイント機構では、チェックポイントバリアが通常のデータレコードよりも優先して処理される必要がある。ストリーム処理の途中でバリアが到着した場合、タスクはすぐにバリアを検出し、アライメント処理やスナップショット取得を開始する必要がある。notifyPriorityはこのような優先度イベントの存在をタスクスレッドに通知し、優先度の高いイベントを即座に処理できるようにする。

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

**通知の受信者**：InputGateのpriorityAvailabilityHelper.getUnavailableToResetAvailable()で取得されるCompletableFutureを待機しているスレッドが受信者となる。通常はタスクのメインスレッドがgetPriorityEventAvailableFuture()を通じてこのFutureを監視している。

**通知内容の概要**：notifyPriority()メソッドは、priorityAvailabilityHelperからCompletableFutureを取得し、そのFutureをcomplete(null)で完了させることで通知を行う。通知自体にはデータは含まれず、「優先度イベントが利用可能になった」という事実のみを伝達する。

**期待されるアクション**：受信者（タスクスレッド）はFutureの完了を検知し、getNext()またはpollNext()でInputGateから優先度イベントを取得する。優先度イベントは通常データより先に処理され、チェックポイントバリアの場合はバリア処理ロジックが起動される。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（CompletableFuture.complete） |
| 優先度 | 最高（優先度イベントの通知であるため） |
| リトライ | なし |

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

InputGateのpriorityAvailabilityHelperが管理するCompletableFutureの待機者に通知される。GateNotificationHelperはclose()メソッドでFutureをcomplete(null)することで通知を実行する。

## 通知テンプレート

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

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

### メソッド定義

```java
/**
 * Must be called under lock to ensure integrity of priorityAvailabilityHelper.
 */
public void notifyPriority() {
    toNotifyPriority = inputGate.priorityAvailabilityHelper.getUnavailableToResetAvailable();
}
```

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| toNotifyPriority | 通知対象のCompletableFuture | inputGate.priorityAvailabilityHelper | Yes |

## 送信トリガー・条件

### トリガー一覧

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

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 優先度要素が既に存在 | inputChannelsWithData.getNumPriorityElements() > 1 の場合 |
| 優先度イベント既にポーリング済み | getPriorityEventAvailableFuture().isDone() == true の場合（UnionInputGate） |
| 古いシーケンス番号 | prioritySequenceNumberが古い場合（SingleInputGate） |

## 処理フロー

### 送信フロー

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

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

### 参照テーブル一覧

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

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

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

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（優先度イベント到着時に即座に通知）

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

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

## 備考

- GateNotificationHelperはAutoCloseableを実装し、try-with-resourcesパターンで使用される
- notifyPriorityで設定されたFutureは、close()メソッドで実際にcomplete()される
- この2段階設計により、ロック内での準備とロック外での通知が分離される
- 優先度イベントの典型例：チェックポイントバリア、EndOfPartitionEvent

---

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

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

### 推奨読解順序

#### 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` | priorityAvailabilityHelperとavailabilityHelperの定義 |

**読解のコツ**: GateNotificationHelperはAutoCloseableであり、try-with-resourcesで使用される。notifyPriority/notifyDataAvailableはロック内で呼び出され、close()でFutureが完了される。

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

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

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

**主要処理フロー**:
1. **27-37行目**: フィールド定義とコンストラクタ
   - **28-29行目**: inputGateとavailabilityMonitorの保持
   - **31-32行目**: toNotifyPriorityとtoNotifyのCompletableFuture
2. **39-47行目**: close()メソッド
   - **41-43行目**: toNotifyPriorityがnullでなければcomplete(null)
   - **44-46行目**: toNotifyがnullでなければcomplete(null)
3. **49-52行目**: notifyPriority()メソッド
   - **51行目**: priorityAvailabilityHelper.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)
  - **1231-1233行目**: 最初の優先度要素の場合にnotification.notifyPriority()

**主要処理フロー（UnionInputGate）**:
- **368-398行目**: queueInputGateメソッド
  - **371-372行目**: try-with-resourcesでGateNotificationHelper作成
  - **373行目**: synchronized (inputGatesWithData)
  - **390-392行目**: 最初の優先度要素の場合にnotification.notifyPriority()

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

```
[ネットワークスレッド等]
        │
        ▼
SingleInputGate.queueChannel(channel, prioritySequenceNumber, forcePriority)
        │
        └─ try (GateNotificationHelper notification = new GateNotificationHelper(...))
               │
               ├─ synchronized (inputChannelsWithData)
               │      │
               │      ├─ queueChannelUnsafe(channel, priority)
               │      │
               │      └─ if (priority && getNumPriorityElements() == 1)
               │             │
               │             └─ notification.notifyPriority()
               │                    │
               │                    └─ toNotifyPriority = priorityAvailabilityHelper
               │                                          .getUnavailableToResetAvailable()
               │
               └─ [ロック解放後] notification.close()
                      │
                      └─ toNotifyPriority.complete(null)
                             │
                             └─ [タスクスレッドのFuture完了]
```

### データフロー図

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

優先度イベント到着 ───▶ queueChannel/queueInputGate ───▶ inputChannelsWithData更新
                               │
                               ▼
                        GateNotificationHelper.notifyPriority()
                               │
                               ▼
                        priorityAvailabilityHelper
                        .getUnavailableToResetAvailable()
                               │
                               ▼
                        [ロック解放後]
                        toNotifyPriority.complete(null)
                               │
                               ▼
                        タスクスレッドのFuture完了通知
                               │
                               ▼
                        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` | クラス | 優先度付きキュー |
