# バッチ設計書 3-AsyncReplicationTask

## 概要

本ドキュメントは、OpenSearchのセグメントレプリケーション対象シャードを定期的に同期するバッチタスクの設計を記載する。

### 本バッチの処理概要

AsyncReplicationTaskは、セグメントレプリケーションまたはリモートストアが有効なインデックスにおいて、検索専用（search-only）レプリカシャードへのセグメントレプリケーションを定期的にトリガーするバッチタスクである。

**業務上の目的・背景**：従来のドキュメントレプリケーション方式では、プライマリシャードとレプリカシャードの両方で独立にインデクシングが行われるが、セグメントレプリケーション方式ではプライマリシャードで生成されたセグメントをそのままレプリカにコピーする。これによりインデクシングのCPU・メモリコストが削減される。本バッチは、このセグメントレプリケーションを定期的にトリガーする役割を担う。

**バッチの実行タイミング**：リフレッシュ間隔（`index.refresh_interval`、デフォルト1秒）と同じ間隔で定期実行される。セグメントレプリケーションまたはリモートストアが有効な場合のみ動作する。

**主要な処理内容**：
1. リフレッシュ間隔が0ms以上であることを確認する
2. インデックス内の全シャードを走査する
3. 検索専用（search-only）かつアクティブなレプリカシャードを特定する
4. 対象シャードに対してreplicator（セグメントレプリケーション処理）を実行する

**前後の処理との関連**：AsyncRefreshTask（No.2）と同じリフレッシュ間隔で動作し、プライマリシャードのリフレッシュ後に新しいセグメントをレプリカへ伝搬する関係にある。AsyncPublishReferencedSegmentsTask（No.4）はプライマリシャードの参照セグメント情報を公開し、本タスクの補完的な役割を持つ。

**影響範囲**：セグメントレプリケーション対象のレプリカシャードに影響する。レプリカの検索可能データの鮮度に直接関わる。

## バッチ種別

データ同期処理（セグメントレプリケーション）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 定期実行（リフレッシュ間隔と同一、デフォルト1秒） |
| 実行時刻 | インデックスのライフサイクルに依存（常時） |
| 実行曜日 | 該当なし（常時稼働） |
| 実行日 | 該当なし（常時稼働） |
| トリガー | タイマーベースの定期スケジュール（AbstractAsyncTask） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| IndexServiceが未クローズ | `indexService.closed.get() == false`であること |
| インデックスがOPEN状態 | `IndexMetadata.State.OPEN`であること |
| セグメントレプリケーション有効 | `indexSettings.isSegRepEnabledOrRemoteNode()`がtrueであること |

### 実行可否判定

`mustReschedule()`がオーバーライドされており、`indexSettings.isSegRepEnabledOrRemoteNode()`がtrueかつBaseAsyncTaskの標準条件を満たす場合のみ再スケジューリングされる。処理本体では`getRefreshInterval().millis() > 0`の場合のみシャード走査が実行される。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| index.refresh_interval | TimeValue | No | 1s | タスクの実行間隔（リフレッシュ間隔を使用） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| IndexServiceのシャードマップ | インメモリ | 対象インデックスに属する全シャード |
| ShardRouting | インメモリ | シャードのルーティング情報（search-only判定に使用） |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| レプリカシャードのセグメント | ディスクファイル | プライマリからコピーされたセグメントファイル |

### 出力ファイル仕様

| 項目 | 内容 |
|-----|------|
| ファイル名 | Luceneセグメントファイル |
| 出力先 | レプリカシャードのインデックスディレクトリ |
| 文字コード | バイナリ形式 |
| 区切り文字 | 該当なし |

## 処理フロー

### 処理シーケンス

```
1. タイマーによる定期起動
   └─ AbstractAsyncTaskによりリフレッシュ間隔で起動
2. リフレッシュ間隔チェック
   └─ getRefreshInterval().millis() > 0 を確認
3. シャード走査
   └─ 全シャードをイテレーション
4. 対象シャード判定
   └─ search-only かつ active なシャードを特定
5. レプリケーション実行
   └─ replicator.accept(shard) でセグメントレプリケーションをトリガー
6. 次回スケジュール
   └─ mustReschedule()がtrueの場合、再スケジュール
```

### フローチャート

```mermaid
flowchart TD
    A[タイマー起動] --> B{リフレッシュ間隔 > 0?}
    B -->|No| G[処理スキップ]
    B -->|Yes| C[全シャード走査開始]
    C --> D{次のシャードあり?}
    D -->|No| H[処理終了・再スケジュール]
    D -->|Yes| E{search-only かつ active?}
    E -->|No| D
    E -->|Yes| F[replicator.accept 実行]
    F --> D
    G --> H
```

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

### 操作別データベース影響一覧

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| セグメントレプリケーション | レプリカシャードのセグメント | WRITE | プライマリのセグメントをレプリカにコピー |

### テーブル別操作詳細

#### レプリカシャードセグメント

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| レプリケーション | セグメントファイル | プライマリシャードの最新セグメント | ファイルベースのレプリケーション |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IndexShardClosedException | シャードが既にクローズされている | 例外をキャッチして処理を継続 |
| - | AlreadyClosedException | リソースが既にクローズされている | 例外をキャッチして処理を継続 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 明示的なリトライなし（次回スケジュール実行で再試行） |
| リトライ間隔 | リフレッシュ間隔（デフォルト1秒） |
| リトライ対象エラー | 次回実行時に自動的に再試行 |

### 障害時対応

レプリケーションに失敗した場合、例外がキャッチされ次のシャードの処理に進む。次回スケジュール実行時に再度レプリケーションが試行される。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | シャード単位（各シャードのレプリケーションは独立） |
| コミットタイミング | レプリケーション完了時 |
| ロールバック条件 | 該当なし |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | search-onlyレプリカシャード数に依存 |
| 目標処理時間 | リフレッシュ間隔内に完了すること |
| メモリ使用量上限 | セグメントコピーに必要なメモリ |

## 排他制御

同一インデックスに対するAsyncReplicationTaskは1つのみ存在する。replicator処理内部でシャードレベルの排他制御が行われる。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| エラーログ | レプリケーション失敗時 | シャードクローズ例外はサイレントにスキップ |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| レプリケーション遅延 | レプリカの同期遅延 | OpenSearchログ |

## 備考

- 本タスクはGENERICスレッドプールで実行される
- セグメントレプリケーションまたはリモートストアが無効な場合、`mustReschedule()`がfalseを返し、タスクは自動的に停止する
- `force`パラメータにtrueを渡すことで強制的にレプリケーションを実行可能だが、定期タスクではforceはfalseで呼ばれる
- ソースコード: `server/src/main/java/org/opensearch/index/IndexService.java` (AsyncReplicationTask内部クラス)
