# 機能設計書 50-Confluent Registry

## 概要

本ドキュメントは、Apache Flink の flink-avro-confluent-registry モジュールが提供する Confluent Schema Registry 連携 Avro シリアライゼーション/デシリアライゼーション機能の設計を記述する。

### 本機能の処理概要

Confluent Registry Avro フォーマット機能は、Confluent Schema Registry と連携して Avro データを読み書きするためのシリアライザとデシリアライザを提供する。スキーマの一元管理とスキーマ進化をサポートする。

**業務上の目的・背景**：Confluent Schema Registry は、Apache Kafka エコシステムで広く使用されるスキーマ管理サービスである。スキーマのバージョン管理、互換性チェック、一元的なスキーマストレージを提供し、データパイプラインの整合性を維持する。特に Kafka を使用したストリーミングデータ処理で不可欠な役割を果たす。

**機能の利用シーン**：
- Kafka メッセージの Avro シリアライゼーション
- スキーマ進化を伴うデータパイプライン
- マイクロサービス間のスキーマ共有
- データ品質の保証
- Debezium CDC データの処理

**主要な処理内容**：
1. RegistryAvroFormatFactory: Schema Registry 連携フォーマットのファクトリ
2. ConfluentRegistryAvroDeserializationSchema: Schema Registry からスキーマを取得してデシリアライズ
3. ConfluentRegistryAvroSerializationSchema: Schema Registry にスキーマを登録してシリアライズ
4. CachedSchemaCoderProvider: スキーマのキャッシュ管理
5. SSL/認証サポート: セキュアな Schema Registry 接続

**関連システム・外部連携**：
- Confluent Schema Registry
- Kafka コネクタ
- Debezium CDC

**権限による制御**：Schema Registry への認証は basic-auth または bearer-auth で設定可能。SSL/TLS による暗号化もサポート。

## 関連画面

本機能はバックエンドのデータフォーマット処理機能であり、直接関連する画面はない。

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

## 機能種別

- データ変換処理
- シリアライゼーション/デシリアライゼーション
- 外部サービス連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| url | String | Yes | Schema Registry URL | 有効な URL |
| subject | String | No* | スキーマサブジェクト | *シリアライズ時は必須 |
| schema | String | No | Avro スキーマ文字列 | 有効な Avro スキーマ |
| properties | Map | No | 追加プロパティ | - |
| ssl.keystore.location | String | No | SSL キーストア場所 | - |
| ssl.keystore.password | String | No | SSL キーストアパスワード | - |
| ssl.truststore.location | String | No | SSL トラストストア場所 | - |
| ssl.truststore.password | String | No | SSL トラストストアパスワード | - |
| basic-auth.credentials-source | String | No | Basic 認証ソース | - |
| basic-auth.user-info | String | No | Basic 認証情報 | - |
| bearer-auth.credentials-source | String | No | Bearer 認証ソース | - |
| bearer-auth.token | String | No | Bearer トークン | - |

### 入力データソース

- デシリアライゼーション: Confluent Wire Format の Avro バイト列
- シリアライゼーション: Flink の RowData 内部データ構造

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| rowData | RowData | Flink 内部データ形式に変換されたデータ |
| avroBytes | byte[] | Confluent Wire Format の Avro バイト列 |

### 出力先

- デシリアライゼーション: Flink の下流オペレーター
- シリアライゼーション: Kafka 等の外部システム

## 処理フロー

### フローチャート

```mermaid
flowchart TD
    subgraph Deserialization
        A[Confluent Wire Format] --> B[Schema ID 抽出]
        B --> C[Schema Registry 問合せ]
        C --> D[Writer Schema 取得]
        D --> E[Avro Deserialize]
        E --> F[RowData]
    end

    subgraph Serialization
        G[RowData] --> H[Avro Serialize]
        H --> I[Schema Registry 登録]
        I --> J[Schema ID 取得]
        J --> K[Confluent Wire Format]
    end
```

### データフロー図

```mermaid
flowchart LR
    subgraph External
        Registry[Schema Registry]
    end

    subgraph Deserialization
        InBytes[byte[]] --> Decoder
        Decoder <--> Registry
        Decoder --> RowData1[RowData]
    end

    subgraph Serialization
        RowData2[RowData] --> Encoder
        Encoder <--> Registry
        Encoder --> OutBytes[byte[]]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-50-01 | URL 必須 | Schema Registry URL は必須 | 全処理 |
| BR-50-02 | シリアライズ時 subject 必須 | シリアライズ時は subject 指定必須 | シリアライゼーション時 |
| BR-50-03 | スキーマキャッシュ | デフォルト 1000 バージョンまでキャッシュ | デシリアライゼーション時 |
| BR-50-04 | スキーマ自動変換 | schema 未指定時はテーブルスキーマから Avro スキーマを自動生成 | スキーマ未指定時 |
| BR-50-05 | プロジェクション対応 | カラムプロジェクションをサポート | 読み取り時 |

### Confluent Wire Format

| 位置 | サイズ | 内容 |
|-----|-------|------|
| 0 | 1 byte | マジックバイト (0x00) |
| 1-4 | 4 bytes | Schema ID (big-endian) |
| 5- | 可変 | Avro バイナリデータ |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ValidationException | subject 未指定 | シリアライズ時に subject 未指定 | subject 設定 |
| IllegalArgumentException | スキーマ不一致 | 指定スキーマとテーブルスキーマの不一致 | スキーマ確認 |
| IOException | Registry 接続エラー | Schema Registry への接続失敗 | URL・認証設定確認 |

## 備考

- Confluent Wire Format に準拠
- GenericRecord と SpecificRecord の両方をサポート
- debezium-avro-confluent フォーマットも提供

---

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

### 推奨読解順序

#### Step 1: ファクトリクラス

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | RegistryAvroFormatFactory.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java` | エントリーポイント |

**主要処理フロー**:
- **84行目**: IDENTIFIER = "avro-confluent"
- **87-121行目**: createDecodingFormat() でデシリアライザ生成
- **109-113行目**: AvroRowDataDeserializationSchema 生成
- **124-161行目**: createEncodingFormat() でシリアライザ生成
- **133-138行目**: subject 未指定時の ValidationException
- **149-153行目**: AvroRowDataSerializationSchema 生成
- **169-173行目**: requiredOptions() で URL 必須指定
- **210-245行目**: buildOptionalPropertiesMap() で認証・SSL プロパティ構築

#### Step 2: 設定オプション

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AvroConfluentFormatOptions.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java` | 設定定義 |

**主要設定**:
- **31-37行目**: URL（必須）
- **39-50行目**: SUBJECT（シリアライズ時必須）
- **52-62行目**: SCHEMA
- **68-90行目**: SSL 設定（keystore/truststore）
- **92-114行目**: 認証設定（basic-auth/bearer-auth）
- **120-128行目**: PROPERTIES（追加プロパティ）

#### Step 3: デシリアライゼーション

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ConfluentRegistryAvroDeserializationSchema.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentRegistryAvroDeserializationSchema.java` | デシリアライザ |

**デシリアライゼーション処理**:
- **42行目**: DEFAULT_IDENTITY_MAP_CAPACITY = 1000
- **74-77行目**: forGeneric() で GenericRecord デシリアライザ生成
- **124-133行目**: forGeneric() フル設定版
- **145-148行目**: forSpecific() で SpecificRecord デシリアライザ生成
- **129-132行目**: CachedSchemaCoderProvider でスキーマキャッシュ

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

```
RegistryAvroFormatFactory
├── createDecodingFormat()
│   ├── ProjectableDecodingFormat
│   │   └── createRuntimeDecoder()
│   │       ├── AvroSchemaConverter.convertToSchema()
│   │       ├── ConfluentRegistryAvroDeserializationSchema.forGeneric()
│   │       │   └── CachedSchemaCoderProvider
│   │       └── AvroRowDataDeserializationSchema
│   └── getChangelogMode() -> insertOnly()
├── createEncodingFormat()
│   ├── EncodingFormat
│   │   └── createRuntimeEncoder()
│   │       ├── AvroSchemaConverter.convertToSchema()
│   │       ├── ConfluentRegistryAvroSerializationSchema.forGeneric()
│   │       └── AvroRowDataSerializationSchema
│   └── getChangelogMode() -> insertOnly()
└── buildOptionalPropertiesMap()
    └── SSL/Auth properties mapping
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| RegistryAvroFormatFactory.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java` | ソース | ファクトリ |
| AvroConfluentFormatOptions.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java` | ソース | 設定オプション |
| ConfluentRegistryAvroDeserializationSchema.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentRegistryAvroDeserializationSchema.java` | ソース | デシリアライザ |
| ConfluentRegistryAvroSerializationSchema.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentRegistryAvroSerializationSchema.java` | ソース | シリアライザ |
| CachedSchemaCoderProvider.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/CachedSchemaCoderProvider.java` | ソース | スキーマキャッシュ |
| ConfluentSchemaRegistryCoder.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java` | ソース | Registry コーダー |
| DebeziumAvroFormatFactory.java | `flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java` | ソース | Debezium 連携 |
