# バッチ設計書 24-BulkProcessor (Flush)

## 概要

本ドキュメントは、OpenSearchのBulkProcessorクラス内部で動作するFlushバッチ処理の設計を記述する。このバッチは、設定されたフラッシュ間隔に基づき、バッファリングされたバルクリクエストを定期的にフラッシュ（送信）する。

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

BulkProcessorは、複数のインデックス・削除リクエストをバッファリングし、一定の条件（アクション数、サイズ、時間間隔）に達したときにまとめてバルクリクエストとして送信するスレッドセーフなプロセッサである。内部のFlushクラスがRunnable実装として定期的に実行され、バッファリングされたアクションがある場合にそれを送信する。

**業務上の目的・背景**：大量のドキュメントをインデクシングする際、1件ずつリクエストを送信するのは非効率であり、ネットワークオーバーヘッドとクラスタ負荷が増大する。BulkProcessorは複数のリクエストをバッファリングしてバルクAPIで一括送信することで効率化を図る。Flushバッチはタイムアウト方式の定期フラッシュを提供し、アクション数やサイズの閾値に達しなくても一定時間経過後にバッファをフラッシュすることで、データの鮮度を保証する。

**バッチの実行タイミング**：Builder.setFlushInterval()で設定されたflushInterval（デフォルトは未設定、null）に基づき、設定された場合のみ定期実行される。

**主要な処理内容**：
1. ロックを取得する
2. BulkProcessorがクローズ済みかどうかを確認する
3. バッファリングされたアクション数が0でないことを確認する
4. バッファリングされたBulkRequestを送信（execute()）し、新しいBulkRequestインスタンスに差し替える

**前後の処理との関連**：アプリケーションがBulkProcessor.add()でドキュメントを追加し、本バッチまたはアクション数/サイズの閾値到達により送信が行われる。BulkRequestHandlerが実際のバルクリクエスト送信とリトライを処理する。

**影響範囲**：BulkProcessorを使用する全てのクライアントコード、Reindexプラグイン、HighLevelRestClient等に影響する。

## バッチ種別

バルク処理（定期フラッシュ）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | flushIntervalに設定された間隔（デフォルト未設定） |
| 実行時刻 | BulkProcessor構築後から継続的に実行 |
| 実行曜日 | 該当なし（常時） |
| 実行日 | 該当なし（常時） |
| トリガー | Scheduler.scheduleWithFixedDelayによる定期実行 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| flushIntervalの設定 | Builder.setFlushInterval()でnullでない値が設定されていること |
| BulkProcessorが未クローズ | closedフラグがfalseであること |

### 実行可否判定

flushIntervalがnullの場合、フラッシュタスクは作成されるがisCancelled=trueのダミーとなる。BulkProcessorがクローズ済みの場合は即座にreturnする。バッファリングされたアクション数が0の場合もreturnする。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| flushInterval | TimeValue | No | null（未設定） | フラッシュ間隔 |
| bulkActions | int | No | 1000 | フラッシュのアクション数閾値 |
| bulkSize | ByteSizeValue | No | 5MB | フラッシュのサイズ閾値 |
| concurrentRequests | int | No | 1 | 同時実行可能なバルクリクエスト数 |
| backoffPolicy | BackoffPolicy | No | exponentialBackoff | リトライのバックオフポリシー |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| BulkRequest（内部バッファ） | Java Object | バッファリングされたIndexRequest/DeleteRequest |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| OpenSearch Bulk API | BulkRequest | バッファリングされたリクエストの一括送信 |

### 出力ファイル仕様

ファイル出力はなし。

## 処理フロー

### 処理シーケンス

```
1. Flushタスクが起動
   └─ Scheduler.scheduleWithFixedDelayにより定期実行
2. ReentrantLockを取得
   └─ バッファ操作のスレッドセーフ確保
3. closedフラグチェック
   └─ trueならreturn
4. バッファリングされたアクション数チェック
   └─ 0ならreturn
5. execute()を実行
   └─ 現在のBulkRequestを送信キューに投入
   └─ 新しいBulkRequestインスタンスを作成し差し替え
6. BulkRequestHandlerが非同期で送信
   └─ concurrentRequestsに基づく同時実行制御
7. ロック解放
```

### フローチャート

```mermaid
flowchart TD
    A[Flushタスク起動] --> B[ロック取得]
    B --> C{BulkProcessorクローズ済み?}
    C -->|Yes| D[return]
    C -->|No| E{バッファアクション数 == 0?}
    E -->|Yes| D
    E -->|No| F[execute: BulkRequestを送信キューに投入]
    F --> G[新しいBulkRequestに差し替え]
    G --> H[ロック解放]
    D --> H
    H --> I[タスク終了]
```

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

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

| 処理 | 対象 | 操作種別 | 概要 |
|-----|------|---------|------|
| バルクリクエスト送信 | OpenSearchインデックス | INDEX/DELETE | バッファリングされたドキュメントの一括書き込み/削除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IllegalStateException | クローズ済みのBulkProcessorへの操作 | "bulk process already closed"メッセージで例外スロー |
| - | InterruptedException | awaitClose中の割り込み | スレッドの割り込みステータスを復元 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | BackoffPolicyに依存（デフォルトはexponential backoff） |
| リトライ間隔 | BackoffPolicyに依存 |
| リトライ対象エラー | スレッドプールがフルの場合等のリソース制約エラー |

### 障害時対応

BulkProcessor.Listenerのコールバック（afterBulk with Throwable）で失敗が通知される。アプリケーション側でエラーハンドリングを実装する必要がある。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | BulkRequest単位 |
| コミットタイミング | バルクリクエスト送信時 |
| ロールバック条件 | 個別ドキュメント単位で成功/失敗が返される |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | バッファ内のアクション数（デフォルト上限1000件） |
| 目標処理時間 | flushInterval内に送信完了すること |
| メモリ使用量上限 | bulkSize（デフォルト5MB） |

## 排他制御

ReentrantLockによりバッファ操作（add/flush/close）が排他制御される。concurrentRequestsパラメータにより、同時に実行されるバルクリクエストの数が制御される（デフォルト1）。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| BulkProcessor.Listener.beforeBulk | バルクリクエスト送信前 | 実行ID、リクエスト内容 |
| BulkProcessor.Listener.afterBulk | 送信成功時 | 実行ID、レスポンス |
| BulkProcessor.Listener.afterBulk | 送信失敗時 | 実行ID、Throwable |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| バルクリクエスト失敗 | Listener.afterBulk(failure)で通知 | アプリケーション側で実装 |
| バッファ滞留 | フラッシュ間隔超過 | アプリケーション側で実装 |

## 備考

- flushIntervalが未設定の場合、時間ベースのフラッシュは無効化される
- bulkActionsとbulkSizeの両方を-1に設定し、flushIntervalのみ設定することで完全な非同期処理が可能
- クローズ時にはバッファ内の残存アクションがフラッシュされる
- Schedulerはflush用とretry用で分離されており、各々独立して管理される
