# 機能設計書 42-Kafkaインジェスト

## 概要

本ドキュメントは、Apache KafkaからOpenSearchへのデータ取り込みを行うKafkaインジェストプラグインの機能設計書である。

### 本機能の処理概要

**業務上の目的・背景**：リアルタイムデータパイプラインにおいて、Apache Kafkaはメッセージブローカーとして広く利用されている。Kafkaインジェストプラグインは、KafkaトピックからOpenSearchインデックスへのプルベースのデータ取り込みを実現し、ストリーミングデータのインデキシングを自動化する。

**機能の利用シーン**：ログデータ、イベントデータ、IoTデータなどがKafkaトピックに流入する環境において、OpenSearchへの自動データ取り込みパイプラインを構築する場面で利用される。

**主要な処理内容**：
1. KafkaConsumerの生成・管理：Kafkaブローカーへの接続とコンシューマーインスタンスの管理
2. パーティション単位のメッセージ読み取り：指定オフセットからのバッチ読み取り
3. オフセット管理：earliest/latest/タイムスタンプベースのオフセット指定
4. ラグ計算：未消費メッセージ数のオフセットベースラグ計算

**関連システム・外部連携**：Apache Kafkaクラスタとの接続が必要。bootstrap_serversによるブローカー指定、トピック名によるデータソース指定を行う。

**権限による制御**：プラグインセキュリティポリシーにより、KafkaConsumerの生成はAccessController.doPrivilegedブロック内で実行される。

## 関連画面

本機能は画面機能マッピングにおいて直接的な画面関連はない（プラグイン経由での操作が主体）。

## 機能種別

データ連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| topic | String | Yes | Kafkaトピック名 | 必須チェック |
| bootstrap_servers | String | Yes | Kafkaブローカーアドレス | 必須チェック |
| auto.offset.reset | String | No | オフセットリセットポリシー | デフォルト: "none" |
| max.poll.records | int | No | 1回のポーリングで取得する最大レコード数 | maxPollSizeのデフォルト値を使用 |
| その他Kafkaコンシューマー設定 | Map<String,Object> | No | Kafkaコンシューマーの追加設定 | - |

### 入力データソース

Apache Kafkaトピックからのメッセージ（キー: byte[], 値: byte[]）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| offset | KafkaOffset (long) | メッセージのKafkaオフセット |
| key | byte[] | メッセージキー |
| payload | byte[] | メッセージペイロード |
| timestamp | long | メッセージタイムスタンプ |

### 出力先

OpenSearchインデックスへのドキュメント登録（IngestionShardConsumerインターフェース経由）

## 処理フロー

### 処理シーケンス

```
1. KafkaPlugin登録
   └─ IngestionConsumerPluginとしてTYPE="KAFKA"でKafkaConsumerFactoryを登録
2. KafkaPartitionConsumer生成
   └─ clientId、KafkaSourceConfig、partitionIdを指定してコンシューマーを生成
3. KafkaConsumer初期化
   └─ ConsumerPropertiesの生成、パーティション割り当て
4. メッセージ読み取り（readNext）
   └─ オフセット指定でseek後、poll()でメッセージをバッチ取得
5. ReadResult変換
   └─ ConsumerRecordをKafkaOffset + KafkaMessageに変換
```

### フローチャート

```mermaid
flowchart TD
    A[KafkaPlugin Loaded] --> B[KafkaConsumerFactory registered]
    B --> C[KafkaPartitionConsumer created]
    C --> D[Kafka Consumer initialized]
    D --> E[Assign partition]
    E --> F{readNext called}
    F --> G[Seek to offset if needed]
    G --> H[consumer.poll]
    H --> I[Convert to ReadResult]
    I --> J[Return results]
    J --> F
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-42-01 | トピック存在チェック | 指定トピックが存在しない場合はIllegalArgumentException | コンシューマー初期化時 |
| BR-42-02 | パーティション存在チェック | 指定パーティションIDがトピックに存在しない場合はIllegalArgumentException | コンシューマー初期化時 |
| BR-42-03 | オフセットリセットデフォルト | auto.offset.resetのデフォルト値は"none"（範囲外オフセットでエラー） | 設定未指定時 |
| BR-42-04 | タイムスタンプベースオフセット | タイムスタンプで対応メッセージが見つからない場合、auto.offset.resetポリシーにフォールバック | pointerFromTimestampMillis呼び出し時 |

### 計算ロジック

ラグ計算: `lag = endOffset - lastFetchedOffset - 1`（getPointerBasedLag メソッド、267-285行目）

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

本機能はKafkaからのデータ読み取りのみを行い、OpenSearchへのドキュメント登録は上位のインジェストフレームワークが担当する。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IllegalArgumentException | トピックが存在しない | トピック名を確認 |
| - | IllegalArgumentException | パーティションが存在しない | パーティションIDを確認 |
| - | IllegalArgumentException | タイムスタンプに対応するメッセージが見つからない（none設定時） | auto.offset.resetをearliestまたはlatestに変更 |
| - | TimeoutException | ポーリングタイムアウト | タイムアウト値の調整またはKafkaブローカーの状態確認 |

### リトライ仕様

上位のインジェストフレームワークによるリトライ制御に依存する。

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

Kafkaコンシューマーはオフセットのコミットを行わず、OpenSearch側でオフセット管理を行う（プルベースインジェスト方式）。

## パフォーマンス要件

- ポーリングタイムアウト: 1000ms（デフォルト、TODO: configurable）
- max.poll.recordsによりバッチサイズを制御可能

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

- KafkaConsumerの生成はAccessController.doPrivilegedブロック内で実行（プラグインセキュリティポリシー適用）
- ClassLoaderの切り替えによるセキュリティコンテキスト制御
- Kafkaブローカーへの認証設定はコンシューマー設定マップ経由で渡す

## 備考

- ByteArrayDeserializerを使用しており、メッセージのキー・値ともにバイト配列で取得される
- max.poll.recordsの設定はコンシューマー初期化時に固定され、readNextのmaxMessagesパラメータは無視される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | KafkaOffset.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaOffset.java` | long型のoffsetフィールドを持つIngestionShardPointer実装 |
| 1-2 | KafkaMessage.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaMessage.java` | key(byte[]), payload(byte[]), timestamp(long)を持つIngestionMessage実装 |
| 1-3 | KafkaSourceConfig.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaSourceConfig.java` | topic, bootstrapServers, autoOffsetResetConfig, maxPollRecordsの設定クラス |

**読解のコツ**: KafkaSourceConfigのコンストラクタ（36-56行目）で、ユーザー設定のオーバーライドとデフォルト値の適用ロジックを確認する。

#### Step 2: エントリーポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KafkaPlugin.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPlugin.java` | IngestionConsumerPluginを実装し、TYPE="KAFKA"でKafkaConsumerFactoryを登録 |
| 2-2 | KafkaConsumerFactory.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaConsumerFactory.java` | KafkaPartitionConsumerを生成するファクトリ |

**主要処理フロー**:
1. **20-41行目（KafkaPlugin）**: プラグイン登録とファクトリ提供

#### Step 3: コンシューマー処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | KafkaPartitionConsumer.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java` | メインのコンシューマー実装。readNext、fetch、ラグ計算を理解する |

**主要処理フロー**:
- **62-90行目**: コンストラクタでトピック・パーティションの存在チェックとassign
- **98-108行目**: createConsumerPropertiesでKafkaコンシューマー設定を生成
- **116-133行目**: createConsumerでPrivilegedActionとしてKafkaConsumerを生成
- **144-155行目**: readNext（オフセット指定）でfetchを呼び出し
- **226-251行目**: fetchメソッドでseek + poll + ReadResult変換
- **267-285行目**: getPointerBasedLagでオフセットベースのラグ計算

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

```
KafkaPlugin (IngestionConsumerPlugin)
    |
    +-- KafkaConsumerFactory (IngestionConsumerFactory)
            |
            +-- KafkaPartitionConsumer (IngestionShardConsumer)
                    |
                    +-- KafkaConsumer<byte[],byte[]> (Kafka Client)
                    |       +-- partitionsFor()
                    |       +-- assign()
                    |       +-- seek()
                    |       +-- poll()
                    |       +-- beginningOffsets()
                    |       +-- endOffsets()
                    |       +-- offsetsForTimes()
                    |
                    +-- KafkaSourceConfig
                    +-- KafkaOffset
                    +-- KafkaMessage
```

### データフロー図

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

Kafka Topic/Partition --> KafkaPartitionConsumer     --> ReadResult<KafkaOffset, KafkaMessage>
  (byte[] key,             |                              (offset, key, payload, timestamp)
   byte[] value,           +-> consumer.poll()
   timestamp)              +-> Convert to KafkaMessage
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KafkaPlugin.java | `plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/` | ソース | プラグインエントリーポイント |
| KafkaConsumerFactory.java | 同上 | ソース | コンシューマーファクトリ |
| KafkaPartitionConsumer.java | 同上 | ソース | パーティション単位のコンシューマー |
| KafkaSourceConfig.java | 同上 | ソース | Kafka接続設定 |
| KafkaOffset.java | 同上 | ソース | オフセットデータ構造 |
| KafkaMessage.java | 同上 | ソース | メッセージデータ構造 |
