# 機能設計書 43-Kinesisインジェスト

## 概要

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

### 本機能の処理概要

**業務上の目的・背景**：AWS環境においてAmazon Kinesisはリアルタイムデータストリーミングサービスとして広く利用されている。KinesisインジェストプラグインはKinesisストリームからOpenSearchインデックスへのプルベースのデータ取り込みを実現し、AWSネイティブなデータパイプラインを構築可能にする。

**機能の利用シーン**：AWS上でKinesis Data Streamsにログ、イベント、メトリクスデータが流入する環境において、OpenSearchへの自動データ取り込みパイプラインを構築する場面で利用される。

**主要な処理内容**：
1. KinesisClientの生成・管理：AWS SDKを使用したKinesisクライアントの初期化
2. シャード単位のレコード読み取り：シーケンス番号ベースのレコードフェッチ
3. シーケンス番号管理：TRIM_HORIZON、LATEST、AT_TIMESTAMP、AT/AFTER_SEQUENCE_NUMBERによるイテレータ制御
4. レコード変換：KinesisレコードからKinesisMessage形式への変換

**関連システム・外部連携**：Amazon Kinesis Data Streamsとの接続が必要。AWS認証情報（access_key, secret_key）、リージョン、ストリーム名の設定が必要。

**権限による制御**：AWS IAM認証によるKinesisストリームへのアクセス制御。静的クレデンシャルプロバイダーを使用。

## 関連画面

本機能は画面機能マッピングにおいて直接的な画面関連はない。

## 機能種別

データ連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| region | String | Yes | AWSリージョン | 必須チェック |
| stream | String | Yes | Kinesisストリーム名 | 必須チェック |
| access_key | String | Yes | AWSアクセスキー | 必須チェック |
| secret_key | String | Yes | AWSシークレットキー | 必須チェック |
| endpoint_override | String | No | エンドポイントオーバーライドURL | デフォルト: "" |

### 入力データソース

Amazon Kinesisストリームからのレコード（byte[]データ + シーケンス番号 + タイムスタンプ）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| sequenceNumber | SequenceNumber (String) | レコードのKinesisシーケンス番号 |
| payload | byte[] | レコードデータ |
| timestamp | Long | 近似到着タイムスタンプ（ミリ秒） |

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. KinesisPlugin登録
   └─ IngestionConsumerPluginとしてTYPE="KINESIS"でKinesisConsumerFactoryを登録
2. KinesisShardConsumer生成
   └─ clientId、KinesisSourceConfig、shardIdを指定
3. KinesisClient初期化
   └─ AWS認証情報、リージョン、エンドポイントを設定
4. ストリーム情報取得
   └─ DescribeStreamでシャード情報を取得し、シャードIDをマッピング
5. レコード読み取り（readNext）
   └─ GetShardIterator → GetRecords でレコードをフェッチ
6. ReadResult変換
   └─ KinesisレコードをSequenceNumber + KinesisMessageに変換
```

### フローチャート

```mermaid
flowchart TD
    A[KinesisPlugin Loaded] --> B[KinesisConsumerFactory registered]
    B --> C[KinesisShardConsumer created]
    C --> D[KinesisClient initialized]
    D --> E[DescribeStream]
    E --> F[Map shard ID]
    F --> G{readNext called}
    G --> H[GetShardIterator]
    H --> I[GetRecords]
    I --> J[Convert to ReadResult]
    J --> K[Return results]
    K --> G
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-43-01 | シャード存在チェック | 指定シャードIDがストリームに存在しない場合はIllegalArgumentException | コンシューマー初期化時 |
| BR-43-02 | レコード上限 | 1回のフェッチで最大10,000レコード（Kinesis API制限） | readNext時 |
| BR-43-03 | ラグ非対応 | ポインタベースのラグ計算は非対応（常に0を返す） | getPointerBasedLag呼び出し時 |

### 計算ロジック

Kinesisはポインタベースのラグ計算をサポートしないため、getPointerBasedLagは常に0を返す（255-257行目）。

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IllegalArgumentException | シャードIDがストリームに存在しない | シャードIDを確認 |
| - | IllegalStateException | シャードイテレータが取得されていない状態でreadNext呼び出し | 初回はオフセット指定でreadNextを呼び出す |
| - | RuntimeException | 無効なエンドポイントオーバーライドURL（URISyntaxException） | URLの構文を確認 |

### リトライ仕様

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

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

Kinesisからのプルベースの読み取りであり、トランザクション管理はOpenSearch側で行う。

## パフォーマンス要件

- Kinesis GetRecords APIの1シャードあたりの制限: 最大10,000レコードまたは10MB
- フェッチ上限: maxMessagesと10,000の小さい方を使用（227行目）

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

- AWS認証にStaticCredentialsProviderを使用（アクセスキー・シークレットキーを直接設定）
- TODO: より安全な認証方式への改善が示唆されている（ソースコード106行目のTODOコメント）
- endpoint_overrideはローカルテスト用途に使用可能

## 備考

- NON_EXISTING_SEQUENCE_NUMBERはシーケンス番号が存在しない場合のセンチネル値
- approximateArrivalTimestampはnullの可能性があるため、null安全な処理が行われている

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SequenceNumber.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/SequenceNumber.java` | String型のsequenceNumberフィールドを持つIngestionShardPointer実装。NON_EXISTING_SEQUENCE_NUMBERセンチネル値を確認 |
| 1-2 | KinesisMessage.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisMessage.java` | payload(byte[]), timestamp(Long)を持つIngestionMessage実装 |
| 1-3 | KinesisSourceConfig.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisSourceConfig.java` | region, stream, accessKey, secretKey, endpointOverrideの設定クラス |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | KinesisPlugin.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisPlugin.java` | TYPE="KINESIS"でKinesisConsumerFactoryを登録 |
| 2-2 | KinesisConsumerFactory.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisConsumerFactory.java` | KinesisShardConsumerを生成するファクトリ |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | KinesisShardConsumer.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/KinesisShardConsumer.java` | メインのコンシューマー実装 |

**主要処理フロー**:
- **64-93行目**: コンストラクタでDescribeStreamを呼び出しシャード情報を取得
- **101-117行目**: createClientでKinesisClientBuilder経由でクライアント生成
- **154-194行目**: fetchRecordsメソッドでGetShardIterator + GetRecordsを実行
- **218-243行目**: fetchメソッドでイテレータタイプを決定しレコードを取得・変換

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

```
KinesisPlugin (IngestionConsumerPlugin)
    |
    +-- KinesisConsumerFactory (IngestionConsumerFactory)
            |
            +-- KinesisShardConsumer (IngestionShardConsumer)
                    |
                    +-- KinesisClient (AWS SDK)
                    |       +-- describeStream()
                    |       +-- getShardIterator()
                    |       +-- getRecords()
                    |
                    +-- KinesisSourceConfig
                    +-- SequenceNumber
                    +-- KinesisMessage
```

### データフロー図

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

Kinesis Stream/Shard --> KinesisShardConsumer           --> ReadResult<SequenceNumber, KinesisMessage>
  (byte[] data,           |                                  (sequenceNumber, payload, timestamp)
   sequenceNumber,        +-> GetShardIterator
   timestamp)             +-> GetRecords
                          +-> Convert to KinesisMessage
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| KinesisPlugin.java | `plugins/ingestion-kinesis/src/main/java/org/opensearch/plugin/kinesis/` | ソース | プラグインエントリーポイント |
| KinesisConsumerFactory.java | 同上 | ソース | コンシューマーファクトリ |
| KinesisShardConsumer.java | 同上 | ソース | シャード単位のコンシューマー |
| KinesisSourceConfig.java | 同上 | ソース | Kinesis接続設定 |
| SequenceNumber.java | 同上 | ソース | シーケンス番号データ構造 |
| KinesisMessage.java | 同上 | ソース | メッセージデータ構造 |
