# バッチ設計書 6-BatchExecLegacySink

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecLegacySinkバッチ実行ノードの設計仕様を記載する。このノードは、レガシーAPI（TableSink）を使用して外部シンクにデータを書き込むバッチ処理を担当する。

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

BatchExecLegacySinkは、Flinkの旧API（TableSink）で定義されたシンクにデータを書き込む実行ノードである。新しいDynamicTableSinkへの移行が推奨されているが、レガシーコネクタとの後方互換性を維持するために存在する。

**業務上の目的・背景**：Flink 1.x系で開発されたレガシーシンクコネクタを使用するアプリケーションとの後方互換性を維持する。新APIへの段階的移行期間中に、既存のTableSinkベースのコネクタを継続利用可能にする。

**バッチの実行タイミング**：Flinkジョブのバッチ実行時に、実行プランの終端ノードとして実行される。上流の全処理が完了した後にデータ書き込みが行われる。

**主要な処理内容**：
1. CommonExecLegacySinkの親クラスメソッドでTransformationを生成
2. 入力行型の検証と変換
3. 出力型クラスのバリデーション（静的・公開・グローバルアクセス可能）
4. TableSinkを使用したデータ出力

**前後の処理との関連**：上流の各種処理ノードからの結果を受け取る。後続処理はなく、データパイプラインの終端となる。

**影響範囲**：レガシーAPIの型システムと新APIの型システム間の変換が発生。型クラスの制約に注意が必要。

## バッチ種別

データシンク（レガシーAPI出力）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | ジョブ実行時（随時） |
| 実行時刻 | ジョブスケジュールに依存 |
| 実行曜日 | 該当なし |
| 実行日 | 該当なし |
| トリガー | Flinkジョブ実行時 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| レガシーテーブルシンク | TableSinkが定義されていること |
| シンク接続 | 外部シンクへの接続が確立可能であること |
| 型制約 | 出力型クラスが静的かつ公開であること |

### 実行可否判定

TableSinkが利用可能であり、出力型が制約を満たすこと。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| tableSink | TableSink<T> | Yes | - | レガシーテーブルシンク |
| upsertKeys | String[] | No | null | アップサートキー |
| inputProperty | InputProperty | Yes | - | 入力プロパティ |
| outputType | LogicalType | Yes | - | 出力型定義 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| Transformation<RowData> | RowData | 上流処理からの入力データ |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| TableSink | レガシーAPI | TableSinkで定義されたシンクへ出力 |

### 出力ファイル仕様

コネクタ実装依存。

## 処理フロー

### 処理シーケンス

```
1. checkAndConvertInputTypeIfNeeded呼び出し
   └─ 入力行型の検証
2. 出力型取得
   └─ tableSink.getConsumedDataType
3. 型バリデーション
   └─ validateTypeで型制約を検証
4. 親クラス処理実行
   └─ CommonExecLegacySink処理
5. シンク出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力行型取得]
    B --> C[出力型取得]
    C --> D[型バリデーション]
    D --> E{型制約OK?}
    E -->|Yes| F[シンク処理実行]
    E -->|No| G[TableException発生]
    F --> H[データ出力]
    H --> I[バッチ終了]
    G --> J[エラー終了]
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| データ書き込み | 外部テーブル | INSERT | レガシーコネクタ経由で書き込み |

### テーブル別操作詳細

コネクタ実装依存。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | 型制約違反 | 型クラスが静的/公開でない | 型クラス定義を修正 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | コネクタ実装依存 |
| リトライ間隔 | コネクタ実装依存 |
| リトライ対象エラー | 一時的な接続エラー |

### 障害時対応

ジョブ失敗時はFlinkのリスタート機構に従い再実行。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | コネクタ実装依存 |
| コミットタイミング | コネクタ実装依存 |
| ロールバック条件 | コネクタ実装依存 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力データ量依存 |
| 目標処理時間 | 並列度・シンク性能依存 |
| メモリ使用量上限 | TaskManager設定に従う |

## 排他制御

並列実行可能。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | シンク開始時 | シンク名 |
| 終了ログ | シンク完了時 | 書き込み件数 |
| エラーログ | エラー発生時 | 例外情報 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 書き込みレコード数/秒 | ジョブ定義依存 | Flink Metrics |

## 備考

- レガシーAPIとの後方互換性のために存在
- 新規開発ではBatchExecSink（DynamicTableSink）の使用を推奨
- needRetraction=false（リトラクション不要）
- isStreaming=false（バッチモード）
- 型クラスは静的かつ公開でグローバルアクセス可能である必要あり
- isMemberClassかつ非staticの場合はエラー
- canonicalNameがnullの場合はエラー
