# 通知設計書 20-GlobalCheckpointListeners通知

## 概要

本ドキュメントは、グローバルチェックポイントが更新された際に登録されたリスナーに新しいチェックポイント値を通知するGlobalCheckpointListeners通知の設計を記述する。

### 本通知の処理概要

GlobalCheckpointListenersは、シャードレベルのグローバルチェックポイントの更新を監視し、リスナーが待機しているチェックポイント値に達した際にリスナーに通知する仕組みである。タイムアウト付きの待機もサポートしており、シャードクローズ時には全リスナーに通知を行う。

**業務上の目的・背景**：OpenSearchのレプリケーションやChange Data Capture（CDC）等の機能では、グローバルチェックポイント（すべてのアクティブレプリカで複製完了が確認されたシーケンス番号）の更新を効率的に検知する必要がある。ポーリングではなくイベント駆動型の通知メカニズムにより、変更の即時検知と低レイテンシの実現が可能になる。

**通知の送信タイミング**：(1) globalCheckpointUpdated()が呼ばれ、新しいチェックポイント値がリスナーの待機値以上の場合。(2) シャードがクローズされた場合（close()メソッド）。(3) リスナーのタイムアウトが発生した場合。

**通知の受信者**：GlobalCheckpointListenerインターフェースを実装するリスナーが受信者である。リスナーはadd()メソッドで登録される。

**通知内容の概要**：更新されたグローバルチェックポイント値（long）と例外情報（Exception、正常更新時はnull）が通知される。シャードクローズ時はUNASSIGNED_SEQ_NOとIndexShardClosedException、タイムアウト時はUNASSIGNED_SEQ_NOとTimeoutExceptionが通知される。

**期待されるアクション**：正常更新時 - リスナーは新しいチェックポイント値を使用して後続の処理（レプリケーション、CDC等）を実行。シャードクローズ時 - リスナーはリソースのクリーンアップを実行。タイムアウト時 - リスナーはリトライまたはエラー処理を実行。

## 通知種別

プロセス内コールバック通知（GlobalCheckpointListenerインターフェースによる非同期通知）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（リスナーのexecutor()で指定されたExecutorで実行） |
| 優先度 | 高（レプリケーション進捗に影響） |
| リトライ | なし（リスナーが再登録する必要あり） |

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

add()メソッドで登録されたリスナーのうち、waitingForGlobalCheckpoint値が現在のグローバルチェックポイント以下のリスナーが通知対象となる。各リスナーは1回のみ通知を受け、通知後はリストから削除される。

## 通知テンプレート

### メール通知の場合

該当なし。

### 本文テンプレート

```
該当なし（long globalCheckpoint + Exception eのコールバック）
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| globalCheckpoint | 更新後のグローバルチェックポイント値 | globalCheckpointUpdated()引数 / UNASSIGNED_SEQ_NO | Yes |
| e | 例外情報（null=正常更新） | null / IndexShardClosedException / TimeoutException | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| チェックポイント更新 | globalCheckpointUpdated(long) | 待機中リスナーの待機値 <= 更新値 | グローバルチェックポイントが更新された場合 |
| シャードクローズ | close() | 全リスナーに通知 | シャードがクローズされた場合 |
| タイムアウト | スケジュール済みタスク実行 | タイムアウト未キャンセルのリスナー | リスナーの待機タイムアウト時 |
| 登録時即時通知 | add() | lastKnownGlobalCheckpoint >= waitingForGlobalCheckpoint | 登録時点で既に条件を満たしている場合 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 待機値未達 | グローバルチェックポイントがリスナーの待機値に達していない場合 |
| タイムアウト後の通知との競合 | タイムアウトにより既にリスナーが削除された場合、タイムアウト処理は発火しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[リスナー登録 add] --> B{シャードクローズ済み?}
    B -->|Yes| C[IndexShardClosedExceptionで即時通知]
    B -->|No| D{lastKnownGCP >= waitingFor?}
    D -->|Yes| E[即時通知 GCP値, null]
    D -->|No| F[listenersマップに登録]
    F --> G{timeout指定あり?}
    G -->|Yes| H[ScheduledFutureでタイムアウト設定]
    G -->|No| I[タイムアウトなしで登録]

    J[globalCheckpointUpdated] --> K[lastKnownGCP更新]
    K --> L[待機値 <= GCPのリスナー抽出]
    L --> M[ScheduledFutureキャンセル]
    M --> N[リスナーに通知 GCP値, null]

    O[close] --> P[closed=true]
    P --> Q[全リスナーに通知 UNASSIGNED_SEQ_NO, IndexShardClosedException]

    H --> R{タイムアウト発生}
    R --> S{リスナーまだ登録中?}
    S -->|Yes| T[リスナー削除]
    T --> U[通知 UNASSIGNED_SEQ_NO, TimeoutException]
    S -->|No| V[何もしない 既に通知済み]
```

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

### 参照テーブル一覧

該当なし（インメモリのグローバルチェックポイント値を使用）。

### 更新テーブル一覧

該当なし。

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| IndexShardClosedException | シャードがクローズされた場合 | 全リスナーにUNASSIGNED_SEQ_NOとIndexShardClosedExceptionで通知 |
| TimeoutException | リスナーの待機がタイムアウト | 対象リスナーにUNASSIGNED_SEQ_NOとTimeoutExceptionで通知 |
| リスナー内の例外 | listener.accept()内で例外発生 | 警告ログを出力（ログメッセージはエラーの種類に応じて分岐） |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（リスナーは1回限りの通知で削除される） |
| リトライ間隔 | なし |
| リトライ対象エラー | なし（リスナーが再登録する必要あり） |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | なし（グローバルチェックポイント更新頻度に依存） |
| 1日あたり上限 | なし |

### 配信時間帯

制限なし。

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

- シャードレベルの内部通知であり、外部への情報漏洩リスクはない
- GlobalCheckpointListenerは@PublicApi(since = "1.0.0")が付与されたパブリックAPIである
- リスナーのexecutor()で指定されたスレッドプールで非同期実行されるため、呼び出し元スレッドのブロックは発生しない

## 備考

- リスナーはLinkedHashMapで管理され、登録順序が保持される
- 各リスナーは最大1回のみ通知を受ける（One-shot pattern）。再度通知を受けたい場合は再登録が必要
- synchronized(this)でスレッド安全性を確保
- assertNotification()でアサーション有効時に通知の整合性を検証
- タイムアウトはScheduledExecutorServiceで管理され、通知時にFutureUtils.cancel()でキャンセル
- lastKnownGlobalCheckpointはUNASSIGNED_SEQ_NO（-2）で初期化される

---

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

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

### 推奨読解順序

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

GlobalCheckpointListenerインターフェースとリスナー管理の仕組みを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | GlobalCheckpointListenerインターフェース（74行目〜）のexecutor()とaccept(long, Exception)メソッド |
| 1-2 | GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | フィールド定義（97-103行目）：closed, listeners(LinkedHashMap), lastKnownGlobalCheckpoint |

**読解のコツ**: GlobalCheckpointListenerはexecutor()メソッドを持ち、通知がどのスレッドで実行されるかをリスナー自身が決定する設計。

#### Step 2: リスナー登録処理を理解する

add()メソッドの即時通知とリスト登録の分岐を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | add()（131行目〜）の3つの分岐：closed時/即時通知/リスト登録+タイムアウト設定 |

**主要処理フロー**:
1. **132-134行目**: closed==trueならIndexShardClosedExceptionで即時通知
2. **136-138行目**: lastKnownGlobalCheckpoint >= waitingForなら即時通知
3. **140-163行目**: リスナーマップに登録、timeoutがあればScheduledFutureを設定

#### Step 3: チェックポイント更新通知を理解する

globalCheckpointUpdated()メソッドの通知フローを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | globalCheckpointUpdated()（199行目〜）と内部のnotifyListeners()（211行目〜） |

**主要処理フロー**:
- **199-209行目**: globalCheckpointUpdated()でlastKnownGlobalCheckpointを更新しnotifyListeners呼び出し
- **220-224行目**: 待機値<=グローバルチェックポイントのリスナーを抽出
- **231-238行目**: タイムアウトFutureをキャンセルし、各リスナーに通知

#### Step 4: 通知実行の詳細を理解する

notifyListener()メソッドのExecutorベース非同期通知を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | notifyListener()（242行目〜）でのexecutor().execute()による非同期通知とエラーハンドリング |

**主要処理フロー**:
- **245行目**: listener.executor().execute()でリスナー指定のExecutorで実行
- **247行目**: listener.accept(globalCheckpoint, e)でコールバック
- **248-261行目**: 例外発生時は種別に応じた警告ログを出力

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

```
IndexShard
    |
    +-- GlobalCheckpointListeners.add(waitingFor, listener, timeout)
    |       |
    |       +-- [closed] notifyListener(UNASSIGNED_SEQ_NO, IndexShardClosedException)
    |       +-- [GCP達成] notifyListener(lastKnownGCP, null)
    |       +-- [待機] listeners.put(listener, (waitingFor, scheduledFuture))
    |
    +-- GlobalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint)
    |       |
    |       +-- lastKnownGlobalCheckpoint = globalCheckpoint
    |       +-- notifyListeners(globalCheckpoint, null)
    |               |
    |               +-- [フィルタ] waitingFor <= globalCheckpoint
    |               +-- FutureUtils.cancel(scheduledFuture)
    |               +-- notifyListener(listener, globalCheckpoint, null)
    |                       |
    |                       +-- listener.executor().execute()
    |                               +-- listener.accept(globalCheckpoint, null)
    |
    +-- GlobalCheckpointListeners.close()
            |
            +-- closed = true
            +-- notifyListeners(UNASSIGNED_SEQ_NO, IndexShardClosedException)
```

### データフロー図

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

リスナー登録               -->  add(waitingFor, listener, timeout)
  waitingForGlobalCheckpoint       |
  timeout                     条件判定

グローバルチェックポイント  -->  globalCheckpointUpdated(gcp)    --> listener.accept(gcp, null)
  更新値                          |                                  (リスナーのexecutor上)
                              notifyListeners(gcp, null)

シャードクローズ           -->  close()                         --> listener.accept(-2, IndexShardClosedException)
                              notifyListeners(-2, exception)

タイムアウト               -->  ScheduledFuture実行             --> listener.accept(-2, TimeoutException)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| GlobalCheckpointListeners.java | `server/src/main/java/org/opensearch/index/shard/GlobalCheckpointListeners.java` | ソース | 通知の主要実装 |
| IndexShard.java | `server/src/main/java/org/opensearch/index/shard/IndexShard.java` | ソース | GlobalCheckpointListenersの保持・呼び出し元 |
| IndexShardClosedException.java | `server/src/main/java/org/opensearch/index/shard/IndexShardClosedException.java` | ソース | シャードクローズ時の例外 |
| SequenceNumbers.java | `server/src/main/java/org/opensearch/index/seqno/SequenceNumbers.java` | ソース | UNASSIGNED_SEQ_NO, NO_OPS_PERFORMED定数 |
| GlobalCheckpointListenersTests.java | `server/src/test/java/org/opensearch/index/shard/GlobalCheckpointListenersTests.java` | テスト | ユニットテスト |
| GlobalCheckpointListenersIT.java | `server/src/internalClusterTest/java/org/opensearch/index/shard/GlobalCheckpointListenersIT.java` | テスト | 統合テスト |
