# 機能設計書 41-コネクタベース

## 概要

本ドキュメントは、Apache Flink の flink-connector-base モジュールが提供するコネクタ開発用基盤クラスの機能設計を記述する。

### 本機能の処理概要

コネクタベース機能は、Flink コネクタ（Source/Sink）を開発するための抽象基盤クラスとユーティリティを提供する。この機能により、外部システムとの接続を行うカスタムコネクタを効率的に実装できる。

**業務上の目的・背景**：Flink で様々な外部システム（Kafka、Kinesis、データベース等）と連携するためには、それぞれのシステムに対応したコネクタが必要である。コネクタベースは、これらのコネクタ開発における共通パターン（スプリット管理、非同期I/O、バッチ処理、レート制限等）を抽象化し、開発者が個別のコネクタ実装に集中できるようにする。これにより、コネクタ開発の生産性向上と品質の均一化を実現する。

**機能の利用シーン**：
- 新規外部システムとの接続コネクタを開発する際
- 既存コネクタをFlinkの新しいSource/Sink APIに移行する際
- 大規模データ処理における非同期シンク実装が必要な際
- 複数ソースを順次切り替えるハイブリッドソースを構築する際

**主要な処理内容**：
1. SourceReaderBase: スプリットベースの並列読み取りを管理する抽象ソースリーダー
2. SplitReader: スプリット単位でのデータ読み取りインターフェース
3. SplitFetcherManager: スプリットフェッチャーのライフサイクル管理
4. AsyncSinkWriter: 非同期バッチ書き込みを行うシンクライターの基盤
5. HybridSource: 複数ソースを順次切り替えるハイブリッドソース
6. RecordEmitter: レコードのダウンストリームへの発行処理

**関連システム・外部連携**：
- Flink Source API（SourceReader, SplitEnumerator）
- Flink Sink API（SinkWriter, StatefulSinkWriter）
- 各種外部システムコネクタ（Kafka, Kinesis, Elasticsearch等）

**権限による制御**：本機能は基盤クラスのため、権限制御は個別のコネクタ実装に委譲される。

## 関連画面

本機能はバックエンドの基盤機能であり、直接関連する画面はない。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

- 基盤機能（Framework）
- データ連携基盤

## 入力仕様

### 入力パラメータ

#### SourceReaderBase

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| splitFetcherManager | SplitFetcherManager | Yes | スプリットフェッチャーの管理クラス | null不可 |
| recordEmitter | RecordEmitter | Yes | レコード発行処理クラス | null不可 |
| config | Configuration | Yes | リーダー設定 | null不可 |
| context | SourceReaderContext | Yes | ソースリーダーコンテキスト | null不可 |
| rateLimiterStrategy | RateLimiterStrategy | No | レート制限戦略 | - |

#### AsyncSinkWriter

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| maxBatchSize | int | Yes | バッチあたりの最大要素数 | > 0 |
| maxBufferedRequests | int | Yes | バッファリングする最大リクエスト数 | > maxBatchSize |
| maxBatchSizeInBytes | long | Yes | バッチあたりの最大バイト数 | > 0 |
| maxTimeInBufferMS | long | Yes | バッファ内の最大滞留時間(ms) | > 0 |
| maxRecordSizeInBytes | long | Yes | 単一レコードの最大バイト数 | <= maxBatchSizeInBytes |
| requestTimeoutMS | long | Yes | リクエストタイムアウト(ms) | > 0 |

### 入力データソース

- ソース側: 外部システムからのデータストリーム（各コネクタ実装依存）
- シンク側: Flinkのデータストリーム

## 出力仕様

### 出力データ

#### SourceReaderBase出力

| 項目名 | 型 | 説明 |
|--------|-----|------|
| records | T | 読み取ったレコード |
| inputStatus | InputStatus | 入力状態（MORE_AVAILABLE, NOTHING_AVAILABLE, END_OF_INPUT） |
| splitStates | List<SplitT> | チェックポイント時のスプリット状態 |

#### AsyncSinkWriter出力

| 項目名 | 型 | 説明 |
|--------|-----|------|
| bufferedRequestState | BufferedRequestState | 未送信リクエストの状態（チェックポイント用） |

### 出力先

- ソース側: Flinkの下流オペレーター
- シンク側: 外部システム（各コネクタ実装依存）

## 処理フロー

### 処理シーケンス

#### SourceReaderBase の処理フロー

```
1. start()
   └─ リーダーの初期化（デフォルトは空実装）

2. addSplits(List<SplitT>)
   └─ 受け取ったスプリットの状態を初期化
   └─ SplitFetcherManagerにスプリットを追加

3. pollNext(ReaderOutput<T>)
   └─ elementsQueueからレコードを取得
   └─ RecordEmitterでレコードを発行
   └─ レート制限が有効な場合は制限を適用

4. snapshotState(checkpointId)
   └─ 全スプリットの現在状態をスナップショット

5. close()
   └─ SplitFetcherManagerをクローズ
```

#### AsyncSinkWriter の処理フロー

```
1. write(InputT, Context)
   └─ ElementConverterで入力をRequestEntryに変換
   └─ バッファに追加
   └─ 必要に応じてnonBlockingFlush()を実行

2. flush(boolean)
   └─ バッファからバッチを作成
   └─ submitRequestEntries()を呼び出し
   └─ 結果をResultHandlerで処理

3. snapshotState(checkpointId)
   └─ 未送信のバッファ内容を状態として返却

4. チェックポイント完了
   └─ 全in-flightリクエストの完了を待機
```

### フローチャート

```mermaid
flowchart TD
    subgraph SourceReaderBase
        A[addSplits] --> B[SplitFetcherManager]
        B --> C[SplitFetcher]
        C --> D[SplitReader.fetch]
        D --> E[elementsQueue]
        E --> F[pollNext]
        F --> G[RecordEmitter.emitRecord]
        G --> H[SourceOutput]
    end

    subgraph AsyncSinkWriter
        I[write] --> J[ElementConverter]
        J --> K[bufferedRequestEntries]
        K --> L{flush条件?}
        L -->|Yes| M[BatchCreator]
        M --> N[submitRequestEntries]
        N --> O[ResultHandler]
        O -->|retry| K
        O -->|complete| P[完了]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-41-01 | バッファサイズ制限 | maxBufferedRequests超過時はflushまでブロック | write()呼び出し時 |
| BR-41-02 | バッチサイズ制限 | maxBatchSize以下のレコード数でバッチ作成 | flush時 |
| BR-41-03 | バイトサイズ制限 | maxBatchSizeInBytes以下でバッチ作成 | flush時 |
| BR-41-04 | タイムアウト処理 | maxTimeInBufferMS経過で自動flush | タイマーコールバック |
| BR-41-05 | レコードサイズ検証 | maxRecordSizeInBytes超過時は例外 | バッファ追加時 |
| BR-41-06 | チェックポイント整合性 | in-flight無し状態でのみスナップショット | snapshotState時 |

### 計算ロジック

**レート制限の適用**:
- RateLimitingStrategyが指定された場合、shouldBlock()でフラッシュ可否を判定
- レコード発行後、rateLimiter.acquire()でパーミッションを取得

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

本機能は基盤クラスのため、直接的なデータベース操作は行わない。データベース操作は個別のコネクタ実装に委譲される。

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | 基盤機能のためDB操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | 入力エラー | maxRecordSizeInBytes超過 | レコードサイズを確認 |
| IllegalStateException | 状態エラー | クローズ後のフェッチャー作成 | ライフサイクル確認 |
| RuntimeException | 実行時エラー | フェッチャースレッドの例外 | 例外内容を確認 |
| TimeoutException | タイムアウト | リクエストタイムアウト | タイムアウト値調整またはリトライ |

### リトライ仕様

**AsyncSinkWriter**:
- submitRequestEntriesの失敗時、ResultHandler.retryForEntries()で再試行
- 失敗エントリはバッファ先頭に戻され、次回flushで再送
- failOnTimeout=falseの場合、タイムアウト時も自動リトライ

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

- SourceReaderBase: チェックポイント機構によりExactly-once保証が可能
- AsyncSinkWriter: At-least-once保証（prepareCommit時に未送信フラッシュ）
- 厳密なトランザクション制御は個別コネクタ実装に委譲

## パフォーマンス要件

- elementsQueueのキャパシティはELEMENT_QUEUE_CAPACITYで設定可能
- 非同期処理によりスループット最大化
- バッチ処理によるネットワークオーバーヘッド削減

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

- 基盤機能のため、認証・認可は個別コネクタ実装に委譲
- fatalExceptionConsumerにより重大例外を適切にハンドリング

## 備考

- このモジュールは@PublicEvolvingアノテーションが付与されており、APIは進化する可能性がある
- HybridSourceを使用する際、最終ソース以外はBOUNDEDである必要がある

---

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

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

### 推奨読解順序

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

まず、ソース基盤とシンク基盤で使用される主要なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | RecordsWithSplitIds.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/RecordsWithSplitIds.java` | スプリットからのレコードを保持するインターフェース |
| 1-2 | BufferedRequestState.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BufferedRequestState.java` | シンクのバッファ状態を保持する構造 |
| 1-3 | RequestEntryWrapper.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/RequestEntryWrapper.java` | リクエストエントリとサイズをラップする構造 |

**読解のコツ**: Flinkのコネクタ基盤は型パラメータが多いため、E（要素型）、T（最終出力型）、SplitT（スプリット型）、SplitStateT（状態型）の違いを意識する。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SourceReaderBase.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java` | ソースリーダーの抽象基底クラス |

**主要処理フロー**:
1. **127-171行目**: コンストラクタでSplitFetcherManager、RecordEmitter、設定を初期化
2. **177-191行目**: pollNext()でレコード取得、レート制限の有無で処理分岐
3. **267-280行目**: getNextFetch()でキューからバッチ取得
4. **303-338行目**: moveToNextSplit()でスプリット間遷移処理
5. **365-374行目**: addSplits()でスプリット追加とフェッチャー起動
6. **348-352行目**: snapshotState()で状態スナップショット

#### Step 3: SplitReaderインターフェースを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SplitReader.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java` | スプリット読み取りインターフェース |

**主要処理フロー**:
- **52行目**: fetch() - スプリットからレコードをフェッチ
- **68行目**: handleSplitsChanges() - スプリット変更の処理
- **71行目**: wakeUp() - ブロッキングfetchの中断

#### Step 4: SplitFetcherManagerを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SplitFetcherManager.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java` | フェッチャースレッドの管理 |

**主要処理フロー**:
- **113-160行目**: コンストラクタでスレッドプール、エラーハンドラー初期化
- **201-230行目**: createSplitFetcher()でフェッチャー生成
- **237-249行目**: maybeShutdownFinishedFetchers()でアイドルフェッチャー終了
- **266-301行目**: close()でグレースフルシャットダウン

#### Step 5: Sink基盤を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | AsyncSinkWriter.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java` | 非同期シンクライターの基盤 |
| 5-2 | ElementConverter.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java` | 入力要素の変換インターフェース |

**主要処理フロー（AsyncSinkWriter）**:
- **241-297行目**: コンストラクタで設定検証、バッファ初期化
- **311-320行目**: write()で要素をバッファに追加
- **353-376行目**: flush()でバッチ作成と送信
- **456-458行目**: snapshotState()でバッファ状態保存
- **480-551行目**: AsyncSinkWriterResultHandlerで非同期結果処理

#### Step 6: HybridSourceを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | HybridSource.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java` | 複数ソース切り替え |

**主要処理フロー**:
- **97-100行目**: コンストラクタでソースリスト保持
- **103-107行目**: builder()でビルダー取得
- **116-119行目**: createReader()でHybridSourceReader生成
- **163-175行目**: SourceSwitchContextで前ソースの状態参照

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

```
SourceReaderBase
    │
    ├─ SplitFetcherManager
    │      ├─ SplitFetcher (スレッド)
    │      │      └─ SplitReader.fetch()
    │      │             └─ RecordsWithSplitIds
    │      └─ FutureCompletingBlockingQueue
    │
    ├─ RecordEmitter.emitRecord()
    │      └─ SourceOutput
    │
    └─ RateLimiter (optional)

AsyncSinkWriter
    │
    ├─ ElementConverter.apply()
    │
    ├─ RequestBuffer (DequeRequestBuffer)
    │      └─ RequestEntryWrapper
    │
    ├─ BatchCreator.createNextBatch()
    │      └─ Batch
    │
    ├─ submitRequestEntries() [abstract]
    │      └─ ResultHandler
    │             ├─ complete()
    │             ├─ completeExceptionally()
    │             └─ retryForEntries()
    │
    └─ RateLimitingStrategy
```

### データフロー図

```
[Source側]
External System ───▶ SplitReader ───▶ RecordsWithSplitIds ───▶ elementsQueue
                                                                    │
                                                                    ▼
                                          SourceOutput ◀─── RecordEmitter

[Sink側]
InputT ───▶ ElementConverter ───▶ RequestEntryT ───▶ bufferedRequestEntries
                                                            │
                                                            ▼
                                  External System ◀─── submitRequestEntries
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SourceReaderBase.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java` | ソース | ソースリーダー抽象基底クラス |
| SplitReader.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java` | ソース | スプリット読み取りインターフェース |
| SplitFetcherManager.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java` | ソース | フェッチャー管理 |
| SplitFetcher.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java` | ソース | 単一フェッチャースレッド |
| RecordEmitter.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/RecordEmitter.java` | ソース | レコード発行インターフェース |
| RecordsWithSplitIds.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/RecordsWithSplitIds.java` | ソース | スプリットレコード保持 |
| FutureCompletingBlockingQueue.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java` | ソース | 非同期通知付きキュー |
| HybridSource.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java` | ソース | ハイブリッドソース |
| HybridSourceReader.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceReader.java` | ソース | ハイブリッドソースリーダー |
| AsyncSinkWriter.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java` | ソース | 非同期シンクライター基盤 |
| AsyncSinkBaseBuilder.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseBuilder.java` | ソース | シンクビルダー基底クラス |
| ElementConverter.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java` | ソース | 要素変換インターフェース |
| BatchCreator.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BatchCreator.java` | ソース | バッチ作成インターフェース |
| RequestBuffer.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/RequestBuffer.java` | ソース | リクエストバッファインターフェース |
| RateLimitingStrategy.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/strategy/RateLimitingStrategy.java` | ソース | レート制限戦略 |
| AIMDScalingStrategy.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/strategy/AIMDScalingStrategy.java` | ソース | AIMDスケーリング戦略 |
| SourceReaderOptions.java | `flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderOptions.java` | ソース | ソースリーダー設定オプション |
