# 通知設計書 20-notifyAvailable

## 概要

本ドキュメントは、Apache Flinkのデータ利用可能通知（階層型ストレージ版、notifyAvailable）の設計仕様を記述したものである。

### 本通知の処理概要

notifyAvailableは、階層型ストレージ（Tiered Storage）において、特定のパーティションとサブパーティションのデータが利用可能になった際にリスナーに通知するメカニズムである。

**業務上の目的・背景**：Flinkの階層型ストレージは、メモリ、ディスク、リモートストレージなど複数の層でデータを管理する。各層でデータが利用可能になったタイミングを効率的に下流タスクに通知することで、低レイテンシのデータ転送を実現する。

**通知の送信タイミング**：TierConsumerAgent（Memory、Disk、Remote）が新しいデータの利用可能性を検知した際に、登録されたAvailabilityNotifierに対して通知が発行される。具体的には、各層のコンシューマエージェントがデータ到着を検知し、通知を行う。

**通知の受信者**：AvailabilityNotifierインターフェースを実装するクラスが受信者となる。主にSingleInputGateやTieredStorageConsumerClientがリスナーとして機能する。

**通知内容の概要**：データが利用可能になったパーティションID（TieredStoragePartitionId）と入力チャネルID（TieredStorageInputChannelId）が通知される。

**期待されるアクション**：受信者は、通知されたパーティションとチャネルからデータを取得する処理を開始する。TieredStorageConsumerClientのgetNextBuffer()メソッドを使用してバッファを取得する。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（データ到着検知スレッドから呼び出し） |
| 優先度 | 高（データ転送に直結） |
| リトライ | なし |

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

TierConsumerAgentに登録されたAvailabilityNotifierに対して通知される。登録はregisterAvailabilityNotifier()メソッドで行われる。

## 通知テンプレート

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

```java
void notifyAvailable(
    TieredStoragePartitionId partitionId,
    TieredStorageInputChannelId inputChannelId);
```

### パラメータ詳細

| パラメータ名 | 型 | 説明 |
|------------|-----|------|
| partitionId | TieredStoragePartitionId | データが利用可能なパーティションのID |
| inputChannelId | TieredStorageInputChannelId | データが利用可能な入力チャネルのID |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| partitionId | パーティション識別子 | TierConsumerAgentから取得 | Yes |
| inputChannelId | 入力チャネル識別子 | TierConsumerAgentから取得 | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| メモリ層 | MemoryTierConsumerAgent | データ利用可能時 | インメモリデータ到着 |
| ディスク層 | DiskTierConsumerAgent | データ利用可能時 | ディスク読み取り完了 |
| リモート層 | RemoteTierConsumerAgent | データ利用可能時 | リモートデータ到着 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 通知者未登録 | AvailabilityNotifierが登録されていない場合 |
| データなし | 該当するデータが存在しない場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[データ到着検知] --> B{どの層?}
    B -->|Memory| C[MemoryTierConsumerAgent]
    B -->|Disk| D[DiskTierConsumerAgent]
    B -->|Remote| E[RemoteTierConsumerAgent]
    C --> F{通知者登録済み?}
    D --> F
    E --> F
    F -->|No| G[処理終了]
    F -->|Yes| H[notifyAvailable呼び出し]
    H --> I[SingleInputGate処理]
    I --> G
```

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

### 参照テーブル一覧

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

### 更新テーブル一覧

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | notifyAvailable内で例外発生 | 例外が上位に伝播 |
| データ取得失敗 | 通知後のデータ取得失敗 | IOException処理 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（データ到着頻度に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

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

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

- 通知はJVM内部のメソッド呼び出しで完結
- パーティションIDとチャネルIDは内部識別子のみ

## 備考

- 階層型ストレージはFlink 1.17以降で導入された新機能
- 複数の層（Memory、Disk、Remote）でデータを効率的に管理

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | TieredStoragePartitionId.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStoragePartitionId.java` | パーティションID構造 |
| 1-2 | TieredStorageInputChannelId.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageInputChannelId.java` | 入力チャネルID構造 |

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

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

**主要処理フロー**:
1. **行28**: インターフェースの目的説明
2. **行30-38**: notifyAvailableメソッドのJavadoc
3. **行37-38**: void notifyAvailable(TieredStoragePartitionId, TieredStorageInputChannelId)

#### Step 3: 登録処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TieredStorageConsumerClient.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java` | 登録処理（149-153行目） |

**主要処理フロー**:
- **行149-153**: registerAvailabilityNotifier()メソッド
- **行150-152**: 全TierConsumerAgentへの登録ループ

#### Step 4: 各層の実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | TierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java` | コンシューマエージェントインターフェース |
| 4-2 | MemoryTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierConsumerAgent.java` | メモリ層実装 |
| 4-3 | DiskTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierConsumerAgent.java` | ディスク層実装 |
| 4-4 | RemoteTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java` | リモート層実装 |

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

```
データ到着（各層）
    │
    ├─ MemoryTierConsumerAgent
    │      │
    │      └─ AvailabilityNotifier.notifyAvailable()
    │
    ├─ DiskTierConsumerAgent
    │      │
    │      └─ AvailabilityNotifier.notifyAvailable()
    │
    └─ RemoteTierConsumerAgent
           │
           └─ AvailabilityNotifier.notifyAvailable()
                  │
                  └─ SingleInputGate
                         │
                         └─ データ取得処理開始
```

### データフロー図

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

データ到着 ───▶ TierConsumerAgent ───▶ AvailabilityNotifier
  (各層)              │                          │
                      ├─ Memory層                └─ SingleInputGate
                      ├─ Disk層                     データ取得
                      └─ Remote層
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AvailabilityNotifier.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/AvailabilityNotifier.java` | インターフェース | 通知インターフェース定義 |
| TieredStorageConsumerClient.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java` | クライアント | 消費者クライアント |
| TierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java` | インターフェース | 層コンシューマエージェント |
| MemoryTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierConsumerAgent.java` | 実装クラス | メモリ層実装 |
| DiskTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/disk/DiskTierConsumerAgent.java` | 実装クラス | ディスク層実装 |
| RemoteTierConsumerAgent.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java` | 実装クラス | リモート層実装 |
| SingleInputGate.java | `flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java` | リスナー実装 | 入力ゲート |
