# 通知設計書 16-notifyBufferAvailable

## 概要

本ドキュメントは、Apache Flinkのバッファ利用可能通知（notifyBufferAvailable）の設計仕様を記述したものである。

### 本通知の処理概要

notifyBufferAvailableは、バッファプールでバッファがリサイクルされ利用可能になった際にリスナーに通知するメカニズムである。バッファ待ちのコンポーネントに対して、新しいバッファが取得可能になったことを知らせる。

**業務上の目的・背景**：ストリーム処理において、メモリバッファは有限のリソースである。バッファプールが枯渇した場合、データ生成側はブロックする必要があるが、バッファが解放されたタイミングで即座に処理を再開するために、この通知メカニズムが使用される。

**通知の送信タイミング**：LocalBufferPoolのrecycle()メソッドが呼び出され、リスナーが登録されている場合に通知が発行される。具体的には、バッファがリサイクルされた際、登録済みのBufferListenerに対してコールバックが実行される。

**通知の受信者**：BufferListenerインターフェースを実装するクラスが受信者となる。主にBufferManager（RemoteInputChannel用）がリスナーとして機能する。

**通知内容の概要**：利用可能になったBufferインスタンスが通知される。受信者はこのバッファを直接使用できる。

**期待されるアクション**：受信者は、通知されたバッファを受け取り、データ受信処理に使用する。notifyBufferAvailableはbooleanを返し、バッファが受け入れられたかどうかを示す。受け入れられなかった場合、バッファはプールに戻される。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（リサイクルスレッドから呼び出し） |
| 優先度 | 高（バッファ管理に直結） |
| リトライ | あり（拒否時は次のリスナーへ） |

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

LocalBufferPoolに登録されたBufferListenerキューから順番にリスナーを取り出し、通知する。リスナーはaddBufferListener()メソッドで登録される。

## 通知テンプレート

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

```java
boolean notifyBufferAvailable(Buffer buffer);
```

### パラメータ詳細

| パラメータ名 | 型 | 説明 |
|------------|-----|------|
| buffer | Buffer | 利用可能になったバッファ |

### 戻り値

| 型 | 説明 |
|-----|------|
| boolean | バッファが受け入れられた場合true、拒否された場合false |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| buffer | 利用可能なバッファ | NetworkBuffer（MemorySegmentをラップ） | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| バッファリサイクル | recycle()メソッド | リスナーが登録されている場合 | バッファ返却時 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| isDestroyed | バッファプールが破棄済みの場合 |
| hasExcessBuffers | 超過バッファがある場合（プールに直接返却） |
| リスナー未登録 | registeredListenersが空の場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[recycle呼び出し] --> B{プール破棄済み?}
    B -->|Yes| C[returnMemorySegment]
    B -->|No| D{超過バッファ?}
    D -->|Yes| C
    D -->|No| E{リスナーあり?}
    E -->|No| F[availableMemorySegmentsに追加]
    E -->|Yes| G[リスナー取得]
    G --> H[fireBufferAvailableNotification]
    H --> I{バッファ受入?}
    I -->|Yes| J[処理完了]
    I -->|No| K{次のリスナーあり?}
    K -->|Yes| G
    K -->|No| F
    F --> J
    C --> J
```

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

### 参照テーブル一覧

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | notifyBufferAvailable内で例外発生 | 例外が上位に伝播（注意が必要） |
| デッドロック | 不適切なロック順序 | ロック外でリスナー呼び出し |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | リスナー数に依存 |
| リトライ間隔 | 即時（同期） |
| リトライ対象エラー | バッファ拒否（falseを返した場合） |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（バッファリサイクル頻度に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

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

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

- 通知はJVM内部のメソッド呼び出しで完結
- バッファの責任がリスナーに移転することに注意

## 備考

- リスナーはバッファを受け取った後、再度登録が必要
- デッドロック回避のため、synchronized外でリスナーを呼び出す

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Buffer.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java` | バッファインターフェース |
| 1-2 | NetworkBuffer.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java` | バッファ実装 |

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

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

**主要処理フロー**:
1. **行25-43**: notifyBufferAvailableメソッドのJavadoc
2. **行43**: boolean notifyBufferAvailable(Buffer buffer)

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | LocalBufferPool.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java` | リサイクル処理（575-609行目） |

**主要処理フロー**:
- **行575-609**: recycle()メソッド
- **行611-618**: fireBufferAvailableNotification()メソッド
- **行651-659**: addBufferListener()メソッド

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

```
バッファリサイクル
    │
    └─ LocalBufferPool.recycle(segment, channel)
           │
           ├─ synchronized(availableMemorySegments)
           │      │
           │      ├─ hasExcessBuffers() チェック
           │      └─ registeredListeners.poll()
           │
           └─ fireBufferAvailableNotification(listener, segment)
                  │
                  └─ BufferListener.notifyBufferAvailable(new NetworkBuffer(...))
                         │
                         └─ BufferManager (RemoteInputChannel)
```

### データフロー図

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

バッファ返却 ───▶ LocalBufferPool ───▶ BufferListener
                       │                          │
                       ├─ リスナーキュー管理       └─ データ受信処理
                       └─ メモリセグメント管理        再開
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| BufferListener.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferListener.java` | インターフェース | リスナーインターフェース定義 |
| LocalBufferPool.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java` | 実装クラス | バッファプール管理 |
| NetworkBuffer.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java` | データ構造 | バッファ実装 |
| BufferManager.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java` | リスナー実装 | バッファ管理 |
| NetworkBufferPool.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java` | グローバルプール | メモリ管理 |
