# 機能設計書 10-Watermark

## 概要

本ドキュメントは、Apache FlinkのWatermark（ウォーターマーク）機能について詳細に記載する。Watermarkはイベント時間処理における進行状況を追跡し、遅延データの処理タイミングを制御する重要な機能である。

### 本機能の処理概要

WatermarkStrategyは、ストリームソースにおいてWatermarkを生成する方法を定義するインターフェースである。Watermarkは「この時刻までのすべてのイベントが到着済み」という宣言であり、イベント時間ウィンドウの完了トリガーやタイマーの発火に使用される。WatermarkGeneratorが実際のWatermark生成を担い、TimestampAssignerがレコードからタイムスタンプを抽出する役割を持つ。

**業務上の目的・背景**：ストリーム処理において、イベントは必ずしもイベント発生順に到着するとは限らない。ネットワーク遅延やシステム障害により、古いイベントが後から到着する可能性がある。Watermarkは、この「順序の乱れ」を許容しつつ、「いつまで待てばウィンドウを確定できるか」を示すメカニズムを提供する。これにより、無限に待つことなく適切なタイミングでウィンドウを発火し、結果を出力できる。また、アイドルソース検知により、データが流れないパーティションがウォーターマークの進行を妨げることを防ぐ。

**機能の利用シーン**：
- イベント時間ウィンドウの完了判定
- 遅延データの許容範囲設定
- アイドルソースの検知
- 複数ソース間のWatermark整列

**主要な処理内容**：
1. WatermarkStrategy：Watermark生成戦略の定義
2. WatermarkGenerator：Watermarkの生成
3. TimestampAssigner：レコードからのタイムスタンプ抽出
4. forMonotonousTimestamps()：単調増加タイムスタンプ用戦略
5. forBoundedOutOfOrderness()：順序乱れを許容する戦略
6. withIdleness()：アイドル検知の追加

**関連システム・外部連携**：SourceコネクタからWatermarkを生成し、オペレーター間でWatermarkを伝播する。WindowOperatorはWatermarkに基づいてイベント時間ウィンドウの発火を判定する。

**権限による制御**：Watermark機能自体には権限制御は存在しない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 17 | ウォーターマーク | 主機能 | ウォーターマーク情報の取得・表示 |

## 機能種別

時系列処理 / イベント時間管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| maxOutOfOrderness | Duration | forBoundedOutOfOrderness | 許容する順序乱れ | >= 0 |
| idleTimeout | Duration | withIdleness | アイドルタイムアウト | > 0 |
| watermarkGroup | String | withWatermarkAlignment | ウォーターマーク整列グループ | Not null |

### 入力データソース

- ソースコネクタからのレコード
- レコードに付与されたタイムスタンプ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Watermark | Watermark | ウォーターマーク値 |
| timestamp | long | 抽出されたタイムスタンプ |

### 出力先

- 下流オペレーター
- WindowOperator（ウィンドウ発火判定）

## 処理フロー

### 処理シーケンス

```
1. WatermarkStrategy設定
   └─ fromSource()またはassignTimestampsAndWatermarks()で設定
2. タイムスタンプ抽出
   └─ TimestampAssigner.extractTimestamp()でレコードからタイムスタンプ取得
3. Watermark生成
   └─ WatermarkGenerator.onEvent()/onPeriodicEmit()でWatermark生成
4. Watermark伝播
   └─ 下流オペレーターにWatermarkを伝播
5. ウィンドウ発火
   └─ Watermarkがウィンドウ終了時刻を超えたらウィンドウ発火
```

### フローチャート

```mermaid
flowchart TD
    A[レコード到着] --> B[TimestampAssigner]
    B --> C[タイムスタンプ抽出]
    C --> D[WatermarkGenerator.onEvent]
    D --> E{周期的emit?}
    E -->|Yes| F[onPeriodicEmit]
    E -->|No| G[次レコード待機]
    F --> H[Watermark出力]
    H --> I[下流へ伝播]
    I --> J{ウィンドウ終了?}
    J -->|Yes| K[ウィンドウ発火]
    J -->|No| G
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 単調増加 | Watermarkは単調増加（減少しない） | 常時 |
| BR-002 | アイドル検知 | 一定時間イベントがないソースをアイドル扱い | withIdleness設定時 |
| BR-003 | 整列 | 複数ソース間でWatermarkを整列 | withWatermarkAlignment設定時 |

### 計算ロジック

**BoundedOutOfOrdernessのWatermark計算**：
```
watermark = maxTimestamp - maxOutOfOrderness - 1
```

## データベース操作仕様

本機能は直接データベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IllegalArgumentException | idleTimeoutが0以下 | 正の値を設定 |
| - | IllegalArgumentException | maxOutOfOrdernessが負 | 0以上を設定 |

### リトライ仕様

チェックポイントが有効な場合、障害時は最後のチェックポイントから再実行。

## トランザクション仕様

Watermark自体にはトランザクション機能はない。

## パフォーマンス要件

- Watermark生成は各レコードまたは周期的に実行されるため軽量実装が推奨
- 周期的emit間隔はautoWatermarkIntervalで設定可能

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

特になし。

## 備考

WatermarkStrategyは@Publicアノテーションが付与された安定したAPIである。

---

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

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

### 推奨読解順序

#### Step 1: 基本インターフェースを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | WatermarkStrategy.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategy.java` | Watermark生成戦略のメインインターフェース |

**読解のコツ**:
- **30-54行目**: JavaDocでWatermarkStrategyの役割と構造を理解
- **56-57行目**: @PublicアノテーションとTimestampAssignerSupplier, WatermarkGeneratorSupplier継承
- **65行目**: createWatermarkGenerator()抽象メソッド
- **71-77行目**: createTimestampAssigner()デフォルト実装（RecordTimestampAssigner）
- **109-133行目**: withTimestampAssigner()メソッド
- **147-153行目**: withIdleness()メソッド
- **219-221行目**: forMonotonousTimestamps()静的メソッド
- **234-236行目**: forBoundedOutOfOrderness()静的メソッド
- **247-249行目**: noWatermarks()静的メソッド

#### Step 2: ビルトイン戦略を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BoundedOutOfOrdernessWatermarks.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/BoundedOutOfOrdernessWatermarks.java` | 順序乱れ許容戦略 |
| 2-2 | AscendingTimestampsWatermarks.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/AscendingTimestampsWatermarks.java` | 単調増加戦略 |

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

```
WatermarkStrategy
    │
    ├─ createWatermarkGenerator(Context)
    │       └─ WatermarkGenerator
    │              ├─ onEvent(event, timestamp, output)
    │              └─ onPeriodicEmit(output)
    │                      └─ WatermarkOutput.emitWatermark()
    │
    └─ createTimestampAssigner(Context)
            └─ TimestampAssigner
                   └─ extractTimestamp(element, recordTimestamp)
```

### データフロー図

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

レコード ───────▶ TimestampAssigner ───────▶ タイムスタンプ付与
     │                   │
     │                   ▼
     │         WatermarkGenerator
     │                   │
     │                   ├─ onEvent()
     │                   └─ onPeriodicEmit()
     │                          │
     │                          ▼
     └───────────────▶ Watermark ────────▶ 下流オペレーター
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| WatermarkStrategy.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategy.java` | ソース | 戦略インターフェース |
| WatermarkGenerator.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkGenerator.java` | ソース | ジェネレーターインターフェース |
| TimestampAssigner.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/TimestampAssigner.java` | ソース | タイムスタンプ抽出 |
| WatermarkOutput.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkOutput.java` | ソース | 出力インターフェース |
| Watermark.java | `flink-core-api/src/main/java/org/apache/flink/api/common/watermark/Watermark.java` | ソース | ウォーターマークオブジェクト |
| BoundedOutOfOrdernessWatermarks.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/BoundedOutOfOrdernessWatermarks.java` | ソース | 順序乱れ許容戦略 |
| AscendingTimestampsWatermarks.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/AscendingTimestampsWatermarks.java` | ソース | 単調増加戦略 |
| WatermarkStrategyWithIdleness.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategyWithIdleness.java` | ソース | アイドル検知付き戦略 |
| RecordTimestampAssigner.java | `flink-core/src/main/java/org/apache/flink/api/common/eventtime/RecordTimestampAssigner.java` | ソース | デフォルトタイムスタンプ抽出 |
