# 通知設計書 15-notifyPriorityEvent

## 概要

本ドキュメントは、Apache Flinkの優先イベント通知（notifyPriorityEvent）の設計仕様を記述したものである。

### 本通知の処理概要

notifyPriorityEventは、バッファキューの先頭に優先イベント（主にチェックポイントバリア）が追加された際にリスナーに通知するメカニズムである。これにより、チェックポイントバリアのような優先度の高いイベントを迅速に処理できる。

**業務上の目的・背景**：Flinkのチェックポイント機構では、データストリームにチェックポイントバリアを挿入し、すべての並列タスクで整合性のある状態スナップショットを作成する。非整合チェックポイント（Unaligned Checkpoint）では、バリアを通常のデータより優先して処理する必要があり、この通知がその制御を担う。

**通知の送信タイミング**：PipelinedSubpartitionに優先度付きバッファ（hasPriority()がtrue）が追加され、かつバッファキューの優先要素数が1（最初の優先イベント）の場合に通知が発行される。具体的にはadd()メソッド内でprocessPriorityBuffer()が呼び出され、needNotifyPriorityEvent()がtrueを返した場合に通知される。

**通知の受信者**：BufferAvailabilityListenerインターフェースを実装するクラス（CreditBasedSequenceNumberingViewReaderなど）が受信者となる。

**通知内容の概要**：優先イベントのシーケンス番号（prioritySequenceNumber）が通知される。これにより受信者は、どの位置に優先イベントがあるかを把握できる。

**期待されるアクション**：受信者は、優先イベントを迅速に読み取り、チェックポイントバリアの処理を開始する。通常のnotifyDataAvailable()と同様に、PartitionRequestQueueへの通知が行われる。

## 通知種別

内部コールバック（Javaメソッド呼び出し）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（バッファ追加スレッドから呼び出し） |
| 優先度 | 最高（チェックポイント処理に直結） |
| リトライ | なし |

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

PipelinedSubpartitionViewの作成時に登録されたBufferAvailabilityListenerに対して通知される。

## 通知テンプレート

### コールバックシグネチャ

```java
default void notifyPriorityEvent(int prioritySequenceNumber) {}
```

### パラメータ詳細

| パラメータ名 | 型 | 説明 |
|------------|-----|------|
| prioritySequenceNumber | int | 優先イベントのシーケンス番号 |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| prioritySequenceNumber | 優先イベントのシーケンス番号 | PipelinedSubpartitionのsequenceNumber | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 優先バッファ追加 | add()メソッド | 最初の優先イベント追加時 | needNotifyPriorityEvent()がtrue |
| バリアタイムアウト | alignedBarrierTimeout() | バリアが優先度に変換された時 | タイムアウト時のバリア昇格 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| isBlocked | サブパーティションがブロック状態の場合 |
| 複数の優先要素 | 既に優先イベントがキューにある場合 |
| DEFAULT_PRIORITY_SEQUENCE_NUMBER | シーケンス番号が-1の場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[優先バッファ追加] --> B[processPriorityBuffer呼び出し]
    B --> C[buffers.addPriorityElement]
    C --> D{チェックポイントバリア?}
    D -->|Yes| E[インフライトバッファ収集]
    E --> F[channelStateWriter.addOutputData]
    D -->|No| G{needNotifyPriorityEvent?}
    F --> G
    G -->|No| H[処理終了]
    G -->|Yes| I[notifyPriorityEvent呼び出し]
    I --> J[readView.notifyPriorityEvent]
    J --> K[BufferAvailabilityListener.notifyPriorityEvent]
    K --> H
```

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

### 参照テーブル一覧

データベースアクセスは発生しない（インメモリ処理）。

### 更新テーブル一覧

データベース更新は発生しない。

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | notifyPriorityEvent内で例外発生 | 例外が上位に伝播 |
| 不正なシーケンス番号 | シーケンス番号が負値 | 処理をスキップ |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0（リトライなし） |
| リトライ間隔 | - |
| リトライ対象エラー | - |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（チェックポイント頻度に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

時間帯の制限なし（24時間稼働）

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

- 通知はJVM内部のメソッド呼び出しで完結
- シーケンス番号は内部制御情報のみ

## 備考

- defaultメソッドとして定義されており、実装は任意
- 非整合チェックポイント（Unaligned Checkpoint）で特に重要

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CheckpointBarrier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/CheckpointBarrier.java` | チェックポイントバリアの構造 |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BufferAvailabilityListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferAvailabilityListener.java` | インターフェース定義（34-39行目） |

**主要処理フロー**:
1. **行34-38**: notifyPriorityEventメソッドのJavadoc
2. **行39**: default void notifyPriorityEvent(int prioritySequenceNumber) {}

#### Step 3: 呼び出し元を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PipelinedSubpartition.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java` | 通知発行の実装（767-772行目） |
| 3-2 | PipelinedSubpartition.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java` | processPriorityBuffer（218-249行目） |

**主要処理フロー**:
- **行767-772**: notifyPriorityEvent()メソッド
- **行252-257**: needNotifyPriorityEvent()判定
- **行196**: add()内でのnotifyPriorityEvent呼び出し

#### Step 4: 受信側を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CreditBasedSequenceNumberingViewReader.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java` | リスナー実装（299-301行目） |
| 4-2 | PipelinedSubpartitionView.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java` | ビュー実装（58-61行目） |

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

```
PipelinedSubpartition.add() [優先バッファ]
    │
    ├─ processPriorityBuffer()
    │      │
    │      ├─ buffers.addPriorityElement()
    │      │
    │      └─ needNotifyPriorityEvent()
    │
    └─ notifyPriorityEvent(prioritySequenceNumber)
           │
           └─ PipelinedSubpartitionView.notifyPriorityEvent()
                  │
                  └─ BufferAvailabilityListener.notifyPriorityEvent()
                         │
                         └─ CreditBasedSequenceNumberingViewReader
                                │
                                └─ notifyDataAvailable() [内部で呼び出し]
```

### データフロー図

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

優先バッファ ───▶ PipelinedSubpartition ───▶ BufferAvailabilityListener
                       │                          │
                       ├─ 優先キュー管理          └─ Netty送信キュー
                       └─ シーケンス番号生成          優先処理
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| BufferAvailabilityListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferAvailabilityListener.java` | インターフェース | リスナーインターフェース定義 |
| PipelinedSubpartition.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java` | 実装クラス | 優先バッファ管理と通知発行 |
| PipelinedSubpartitionView.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java` | 実装クラス | ビュー実装 |
| CreditBasedSequenceNumberingViewReader.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java` | リスナー実装 | Netty層での処理 |
| CheckpointBarrier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/CheckpointBarrier.java` | データ構造 | チェックポイントバリア |
