# 通知設計書 32-onNotification

## 概要

本ドキュメントは、Apache Flinkにおける汎用通知コールバック（onNotification）の設計仕様を記述したものである。

### 本通知の処理概要

onNotificationは、NotificationListenerインターフェースで定義された汎用的な通知コールバックメソッドである。主に非同期I/O操作の完了通知に使用され、特にAsynchronousFileIOChannelにおいて、すべての保留中のI/Oリクエストが処理完了した際に登録されたリスナーに通知する役割を担う。

**業務上の目的・背景**：Flinkの分散ストリーム処理では、大量のデータをディスクにスピルアウトする際に非同期I/Oを使用する。非同期I/O操作は複数のリクエストがキューに投入され、バックグラウンドのワーカースレッドによって処理される。この通知機能により、呼び出し側は全てのI/Oリクエストが完了したことを非同期に検知でき、リソースのクリーンアップやチャネルのクローズなどの後続処理を適切なタイミングで実行できる。

**通知の送信タイミング**：AsynchronousFileIOChannelにおいて、requestsNotReturnedカウンターが0になった時点（すべての未処理リクエストが完了した時点）で通知が送信される。これは、handleProcessedBuffer()メソッド内で各I/Oリクエストの完了時にカウンターがデクリメントされ、カウンターが0になった際にリスナーに通知される。

**通知の受信者**：registerAllRequestsProcessedListener()メソッドで登録されたNotificationListenerインスタンスが受信者となる。リスナーは1つのみ登録可能で、重複登録を試みるとIllegalStateExceptionがスローされる。

**通知内容の概要**：onNotification()メソッドはパラメータを持たず、単に「すべてのリクエストが処理完了した」という事実のみを通知する。具体的な処理結果やエラー情報は別途RequestDoneCallbackを通じて伝達される。

**期待されるアクション**：受信者は通知を受けて、チャネルのクローズ処理や次のバッチ処理の開始など、I/O完了後の後続処理を実行することが期待される。ただし、通知直後に新しいリクエストが到着する可能性があるため、チャネルがクローズされていない限り、未処理リクエスト数が再び増加する可能性がある点に注意が必要である。

## 通知種別

内部コールバック通知（Java インターフェースベースのリスナーパターン）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（I/Oワーカースレッドから直接呼び出し） |
| 優先度 | 高 |
| リトライ | なし |

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

AsynchronousFileIOChannel内のallRequestsProcessedListenerフィールドに登録されている単一のNotificationListenerに通知される。リスナーは登録後、通知が行われると自動的にnullにリセットされる（1回限りの通知）。

## 通知テンプレート

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

| 項目 | 内容 |
|-----|------|
| インターフェース | NotificationListener |
| メソッドシグネチャ | void onNotification() |

### メソッド定義

```java
public interface NotificationListener {
    void onNotification();
}
```

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| なし | パラメータなしのコールバック | - | - |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| I/O完了 | handleProcessedBuffer() | requestsNotReturned.decrementAndGet() == 0 | 最後のI/Oリクエストが処理完了 |
| チャネルクローズ | addRequest() | closed && listener != null | クローズ済みチャネルへのリクエスト追加時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| リスナー未登録 | allRequestsProcessedListenerがnullの場合 |
| 未処理リクエスト残存 | requestsNotReturned > 0 の場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[I/Oリクエスト完了] --> B[handleProcessedBuffer]
    B --> C[resultHandler.requestSuccessful/Failed]
    C --> D{closeLock同期ブロック}
    D --> E[requestsNotReturned.decrementAndGet]
    E --> F{カウンター == 0?}
    F -->|No| G[終了]
    F -->|Yes| H{チャネルクローズ中?}
    H -->|Yes| I[closeLock.notifyAll]
    I --> J{リスナー登録あり?}
    H -->|No| J
    J -->|Yes| K[listener = allRequestsProcessedListener]
    K --> L[allRequestsProcessedListener = null]
    L --> M[listener.onNotification]
    J -->|No| G
    M --> G
```

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

### 参照テーブル一覧

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

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| IllegalStateException | 既にリスナーが登録されている状態で再度登録を試みた場合 | 例外をスローし処理を中断 |
| 登録失敗（false返却） | 登録時に既に未処理リクエストが0の場合 | registerメソッドがfalseを返す |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（I/Oリクエスト完了時に即座に通知）

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

- 通知はJVMプロセス内の内部通信であり、ネットワーク経由では送信されない
- リスナーの登録と通知はロック（listenerLock）で保護されている
- onNotification()はcloseLockの外で呼び出されるため、デッドロックのリスクを軽減

## 備考

- 通知は1回限りで、通知後にリスナーはnullにリセットされる
- 新しいリクエストが到着する可能性があるため、通知時点でリクエスト数が0であることは保証されない
- チャネルクローズ中の場合、closeLock.notifyAll()も同時に呼び出される
- BufferFileWriterなど、AsynchronousFileIOChannelのサブクラスで利用される

---

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

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

### 推奨読解順序

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

まず、通知インターフェースの定義を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | NotificationListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java` | 汎用通知リスナーインターフェース。パラメータなしのonNotification()メソッドのみを定義 |

**読解のコツ**: このインターフェースは非常にシンプルで、パラメータを持たない。通知の詳細な情報は、このインターフェース以外の手段で伝達される設計である。

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

I/Oチャネルの基底クラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AsynchronousFileIOChannel.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousFileIOChannel.java` | 非同期I/Oチャネルの実装。リスナー管理と通知送信ロジック |

**主要処理フロー**:
1. **74行目**: allRequestsProcessedListenerフィールドの定義
2. **196-231行目**: handleProcessedBuffer()メソッド
   - **215行目**: requestsNotReturned.decrementAndGet()
   - **216-218行目**: クローズ中の場合のnotifyAll()
   - **220-223行目**: リスナーの取得とnullリセット
   - **227-229行目**: listener.onNotification()の呼び出し
3. **233-260行目**: addRequest()メソッド
   - **245-254行目**: クローズ済みの場合のリスナー通知
4. **272-290行目**: registerAllRequestsProcessedListener()メソッド
   - **279-280行目**: 既に未処理リクエストが0の場合はfalseを返す
   - **283行目**: リスナーの登録

#### Step 3: 具体的な利用例を理解する

バッファファイルライターでの利用を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | BufferFileWriter.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriter.java` | NotificationListenerの具体的な利用インターフェース |
| 3-2 | AsynchronousBufferFileWriter.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousBufferFileWriter.java` | BufferFileWriterの実装クラス |

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

```
[I/Oワーカースレッド]
        │
        ▼
AsynchronousFileIOChannel.handleProcessedBuffer(T buffer, IOException ex)
        │
        ├─ resultHandler.requestSuccessful(buffer) または requestFailed(buffer, ex)
        │
        └─ synchronized (closeLock)
               │
               ├─ requestsNotReturned.decrementAndGet()
               │
               └─ if (count == 0)
                      │
                      ├─ if (closed) closeLock.notifyAll()
                      │
                      └─ synchronized (listenerLock)
                             │
                             ├─ listener = allRequestsProcessedListener
                             │
                             └─ allRequestsProcessedListener = null
                                    │
                                    └─ listener.onNotification()
                                           │
                                           └─ [呼び出し側の完了処理]
```

### データフロー図

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

I/Oリクエスト ─────▶ RequestQueue ─────▶ I/Oワーカースレッド
                                                │
                                                ▼
                                        handleProcessedBuffer()
                                                │
                                                ▼
                                        requestsNotReturned--
                                                │
                                        ┌───────┴───────┐
                                        │ count == 0?   │
                                        └───────┬───────┘
                                                │Yes
                                                ▼
                                        listener.onNotification()
                                                │
                                                ▼
                                        [完了処理実行]
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| NotificationListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java` | インターフェース | 汎用通知リスナーインターフェース |
| AsynchronousFileIOChannel.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousFileIOChannel.java` | 抽象クラス | 非同期I/Oチャネルの基底クラス、リスナー管理と通知送信 |
| BufferFileWriter.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriter.java` | インターフェース | バッファファイルライターのインターフェース |
| AsynchronousBufferFileWriter.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousBufferFileWriter.java` | クラス | 非同期バッファファイルライターの実装 |
| TestNotificationListener.java | `flink-runtime/src/test/java/org/apache/flink/runtime/io/network/util/TestNotificationListener.java` | テスト | テスト用のNotificationListener実装 |
| AsynchronousBufferFileWriterTest.java | `flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousBufferFileWriterTest.java` | テスト | 非同期バッファライターのテスト |
| AsynchronousFileIOChannelTest.java | `flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/AsynchronousFileIOChannelTest.java` | テスト | 非同期I/Oチャネルのテスト |
