# バッチ設計書 25-SegmentReplicationPressureService

## 概要

本ドキュメントは、OpenSearchのSegmentReplicationPressureService内部で動作するAsyncFailStaleReplicaTaskバッチ処理の設計を記述する。このバッチはセグメントレプリケーション環境において遅延したレプリカシャードを定期的に検出し、一定条件を満たした場合にそのレプリカを失敗（fail）させる。

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

SegmentReplicationPressureServiceは、セグメントレプリケーションが有効な環境で、プライマリシャードに対するレプリカシャードの同期遅延を監視する。内部のAsyncFailStaleReplicaTaskが30秒間隔で動作し、レプリケーション時間の制限（segrep.replication.time.limit）を超過したレプリカシャードをShardStateAction.remoteShardFailed()によって失敗状態にする。

**業務上の目的・背景**：セグメントレプリケーション方式では、プライマリシャードのセグメントをレプリカにコピーする。レプリカが大幅に遅延すると、プライマリのリソースが無駄に消費され、クラスタ全体の安定性が損なわれる。本バッチは遅延レプリカを自動検出・排除することで、クラスタの健全性を維持する。また、バックプレッシャー機能（segrep.pressure.enabled）と組み合わせることで、遅延時にプライマリへの書き込みを拒否する制御も行う。

**バッチの実行タイミング**：30秒固定間隔でAbstractAsyncTaskにより定期実行される。ただし、segrep.replication.time.limitが0（デフォルト、無効）の場合はリスケジュールされない。

**主要な処理内容**：
1. SegmentReplicationStatsTrackerから全シャードのレプリケーション統計を取得する
2. 各シャードの遅延レプリカを特定する（チェックポイント超過 + レプリケーション時間超過）
3. 最もレプリケーション時間が長いシャードを選択する
4. そのシャードの遅延レプリカに対してremoteShardFailed()を呼び出す

**前後の処理との関連**：TransportShardBulkActionがisSegrepLimitBreached()を呼び出してバックプレッシャーチェックを行う。ShardStateActionがレプリカの失敗処理を行い、クラスタ状態が更新される。

**影響範囲**：セグメントレプリケーションが有効なインデックスの全プライマリシャードとそのレプリカに影響する。

## バッチ種別

レプリケーション管理・異常検出

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 30秒固定 |
| 実行時刻 | サービス起動後から継続的に実行 |
| 実行曜日 | 該当なし（常時） |
| 実行日 | 該当なし（常時） |
| トリガー | AbstractAsyncTaskによる自動リスケジュール |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| segrep.replication.time.limit | 0以外（デフォルト0分=無効）に設定されていること |
| セグメントレプリケーション有効 | 対象インデックスでセグメントレプリケーションまたはリモートノードが有効であること |

### 実行可否判定

mustReschedule()はshouldScheduleAsyncFailTask()を呼び出し、replicationTimeLimitFailReplicaがTimeValue.ZEROでない場合にtrueを返す。runInternal()内でも同じ条件をチェックし、無効の場合はスキップする。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| segrep.pressure.enabled | boolean | No | false | バックプレッシャーの有効/無効 |
| segrep.pressure.checkpoint.limit | int | No | 30 | チェックポイント遅延数の閾値 |
| segrep.pressure.time.limit | TimeValue | No | 5m | バックプレッシャー発動のレプリケーション時間閾値 |
| segrep.replication.time.limit | TimeValue | No | 0m（無効） | レプリカ失敗のレプリケーション時間閾値 |
| segrep.pressure.replica.stale.limit | double | No | 0.5 | staleレプリカの割合上限 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| SegmentReplicationStatsTracker | Java API | シャードごとのレプリケーション統計 |
| SegmentReplicationShardStats | Java Object | レプリカ単位のレプリケーション遅延情報 |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| ShardStateAction.remoteShardFailed() | Java API | 遅延レプリカシャードの失敗通知 |

### 出力ファイル仕様

ファイル出力はなし。

## 処理フロー

### 処理シーケンス

```
1. shouldScheduleAsyncFailTask()でタスク有効性を確認
   └─ replicationTimeLimitFailReplicaが0でなければ有効
2. SegmentReplicationStatsTrackerからシャード統計を取得
3. 各シャードの遅延レプリカをフィルタリング
   └─ checkpointsBehindCount > maxCheckpointsBehind AND currentReplicationTimeMillis > replicationTimeLimitBackpressure
4. 最もレプリケーション時間が長いシャードIDを選択
   └─ Comparator.comparingLong で最大値を取得
5. 選択されたシャードの遅延レプリカに対してループ
6. currentReplicationTimeMillis > replicationTimeLimitFailReplica のレプリカを失敗させる
   └─ ShardStateAction.remoteShardFailed()を呼び出し
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B{replicationTimeLimitが有効?}
    B -->|No| C[スキップ]
    B -->|Yes| D[レプリケーション統計取得]
    D --> E[遅延レプリカのフィルタリング]
    E --> F{遅延レプリカあり?}
    F -->|No| G[バッチ終了]
    F -->|Yes| H[最大遅延シャード選択]
    H --> I[遅延レプリカに対してループ]
    I --> J{時間制限超過?}
    J -->|Yes| K[remoteShardFailed呼び出し]
    J -->|No| I
    K --> I
    I --> G
    C --> G
```

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

本バッチはデータベースを使用しない。クラスタ状態の更新はShardStateActionを通じて行われる。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | Exception | remoteShardFailed()の失敗 | ERRORログ出力 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（次回30秒後に再チェック） |
| リトライ間隔 | 30秒 |
| リトライ対象エラー | 全てのエラー（次回実行で自動リトライ） |

### 障害時対応

個別のレプリカ失敗処理のエラーはERRORログに記録される。バッチ全体の処理は継続される。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | なし |
| コミットタイミング | 該当なし |
| ロールバック条件 | 該当なし |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | セグメントレプリケーション対象シャード数に依存 |
| 目標処理時間 | 30秒の間隔内に完了すること |
| メモリ使用量上限 | 軽量（統計データの集計程度） |

## 排他制御

1回のバッチ実行で最も遅延の大きいシャードのみを処理するため、段階的な負荷軽減を行う設計となっている。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| WARNログ | バックプレッシャーによる書き込み拒否時 | 遅延レプリカの割合とシャード情報 |
| TRACEログ | remoteShardFailed成功時 | シャードIDとアロケーションID |
| ERRORログ | remoteShardFailed失敗時 | エラー詳細 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| レプリカ失敗処理の実行 | 発生時 | OpenSearchログ |
| 書き込み拒否 | 発生時 | OpenSearchログ（WARNレベル） |

## 備考

- バッチ間隔は30秒固定でAsyncFailStaleReplicaTask.INTERVAL定数として定義されている
- 1回の実行で1つのシャードIDのレプリカのみを処理する（段階的な負荷低減）
- segrep.replication.time.limitが動的に変更された場合、タスクは再作成される（updateAsyncFailReplicaTask）
- GENERICスレッドプールで実行される
