kafka-schema-registry

Kafka Schema Registry スキル

プロジェクトをスキャンして Kafka アプリケーションを特定し、スキーマを抽出して、Schema Registry への登録用 Terraform を生成し、包括的な分析レポートを作成します。

使用するタイミング

次の場合にこのスキルを呼び出します：

• イベントスキーマの追加や Schema Registry との統合に向けて、Kafka の使用状況をプロジェクトで分析したい場合
• Kafka プロデューサーからスキーマを抽出したい場合
• Schema Registry にスキーマを登録するための Terraform を作成したい場合
• Kafka プロデューサー／コンシューマーの設定を監査したい場合

成果物

このスキルは、対象プロジェクトに次の 3 つの出力を生成します：

1. schema-report.md — 分析結果、リスク、アップグレードの推奨事項を含む完全なレポート
2. schemas/ — PII タグ付きの抽出済みスキーマファイル（Avro、JSON Schema、Protobuf）
3. terraform/ — Confluent プロバイダーを使用してスキーマを登録する Terraform 設定

オプション：コード移行の支援

アプリケーションコードを更新して Schema Registry と統合するようユーザーから依頼された場合は、コード移行リファレンスを使用し、適切な Schema Registry 統合パターンに沿ってコードを更新します。

---

大まかなワークフロー

フェーズ 0：初期化

• 既存の schema.yaml と schemas/ ディレクトリがあるか手動で確認します
• 既存のスキーマ基盤があれば、レポートに記載します

フェーズ 1：プロジェクトのスキャンと Kafka の検出

1. ビルドファイルを検索する — pom.xml、build.gradle、requirements.txt、package.json などを検索します
2. Kafka 依存関係を検出する — spring-kafka、confluent-kafka、kafkajs などを探します
3. プロデューサーとコンシューマーを特定する — KafkaTemplate、Producer(、producer.send などを Grep で検索します
4. トピック名を抽出する — 文字列リテラル、設定プロパティ、YAML ファイルから抽出します
5. シリアライザーを特定する — value.serializer、KafkaAvroSerializer、カスタムシリアライザーを探します
6. アプリケーションカタログを作成する — アプリ名、言語、役割、トピック、シリアライザー、カテゴリをまとめます

詳細なパターン： 検出パターンのリファレンス

アプリケーションカタログの構造：

app_name: module name
language: Java | Python | .NET | Go | Node/TS
role: producer | consumer | both
topics: [list of topics]
serializer_class: value.serializer used
custom_serializer: true | false
schema_format: AVRO | JSON | PROTOBUF | UNKNOWN
sr_integrated: true | false
category: A | B | C | D | E  # REQUIRED

複数スキーマを持つトピックの検出：

• 複数のデータモデルが同じトピックに出力する場合は、oneOf／union／oneof を含むラッパースキーマを作成します
• schema_reference ブロックを含む Terraform を生成します
• レポートで目立つように注意喚起します

フェーズ 2：リスクの検出

次を検索します：

• auto.register.schemas=true — 制御されていないスキーマ進化（カテゴリ C）
• use.latest.version — 設定すると移行が容易になります
• カスタムシリアライザー — SR を完全にバイパスします（カテゴリ E）

該当する各箇所について、ファイルパス、行番号、影響を受けるトピックを記録します。

パターン： 検出パターンのリファレンス

フェーズ 3：スキーマの推論

各プロデューサーについて、次の作業を行います：

1. 既存のスキーマファイルを確認する — **/*.avsc、**/*.proto、**/*.schema.json
2. データモデルから推論する — Java クラス、Pydantic モデル、TypeScript インターフェース、Go 構造体
3. インラインデータから推論する — HashMap、dict リテラル、map[string]any、プレーンオブジェクト、JSON 文字列
4. スキーマに変換する — 言語の型を JSON Schema／Avro／Protobuf に対応付けます
5. PII フィールドにタグを付ける — フィールド名から email、ssn、phone、address などを検索します

PII タグ付け： 検出したフィールドに confluent:tags（PII、PRIVATE、SENSITIVE、PHI）を追加します。

詳細な推論パターン： スキーマ推論のリファレンス

フェーズ 4：プロデューサーを分類する

各プロデューサーを分類します。

カテゴリ	基準
A: 準拠	Confluent シリアライザー + SR + auto.register なし
A→Header	すでに SR を使用しており、ヘッダーへ移行中
B: コード内にスキーマ、SR なし	データモデルは存在するが、SR との統合がない
C: 自動登録	auto.register.schemas=true
D: スキーマなし	生の文字列／バイト列、データモデルなし
E: カスタムシリアライザー	カスタムの Serializer<T>、または SR を使用しないインラインシリアライズ

重要： 次の箇所では、正確な表現「Category X」を使用してください。

• アプリカタログのフィールド
• Applications Discovered テーブル
• レポートのセクション見出し
• Terraform のコメント
• リスクのセクション

詳細： 分類リファレンス

フェーズ 5：スキーマファイルを作成する

ディレクトリ構成：

schemas/
├── avro/
│   └── {topic}-value.avsc
├── json/
│   └── {topic}-value.json
└── proto/
    └── {topic}-value.proto

ファイル名： 必ず kebab-case（小文字とハイフン）を使用してください。

• Value：{topic}-value.{ext}
• Key：{topic}-key.{ext}
• 例：order-events-value.avsc、user-notifications-value.json

初期化： schema.yaml を作成します。

検証： 利用可能な場合は schema_lint(path: schemas/, fix: true) を呼び出します。

フェーズ 6：Terraform を生成する

ファイル構成（個別のファイルに分けることが必須）：

terraform/
├── providers.tf              # Provider config
├── variables.tf              # Variable definitions
├── tags.tf                   # confluent_tag resources (if PII exists)
├── schemas.tf                # Active schemas (A, B, E)
├── flagged-auto-register.tf  # Category C only (commented out)
├── outputs.tf                # Output values
└── import.sh                 # Import script

重要：

• schemas.tf = カテゴリ A、B、E — コメントアウトしないでください
• flagged-auto-register.tf = カテゴリ C のみ — 必ずコメントアウトしてください
• confluent:tags を使用するスキーマがある場合、tags.tf は必須です
• 各スキーマリソースに、Topic、App、Source、Category を含むコメントブロックが必要です

テンプレート： Terraform テンプレートリファレンス

フェーズ 7：レポートを生成する

次の内容を含む schema-report.md を作成します。

• エグゼクティブサマリー（指標 + カテゴリ別内訳）
• Applications Discovered テーブル（形式を厳守し、Category 列は必須）
• RISKS（自動登録、カスタムシリアライザー）
• プロデューサーのアップグレードに関する推奨事項（アプリごとに、「Category X」を見出しに含める）
• 移行のロールアウト順序（カテゴリ別）
• 検出された PII フィールド
• 生成された Terraform リソース
• 次のステップのチェックリスト

重要な書式要件：

1. Applications Discovered は Markdown テーブルにしてください。説明形式のセクションにはしないでください
2. すべてのアプリのセクションに「Category X」を明記してください
3. すべてのリソースに Terraform のコメントブロックが必要です

テンプレート： レポートテンプレートリファレンス

---

カテゴリ別の移行ロールアウト

• Category B（JSON、SR なし）：プロデューサーを先に移行し、その後コンシューマーを移行
• Category A→Header（すでに SR を使用）：コンシューマーのバージョンを確認し、プロデューサーのみ移行
• Category C（自動登録）：Terraform 経由で登録 → 自動登録を無効化 → プロデューサーが最新を取得
• Category E（カスタムシリアライザー）：コンシューマーを先に移行（複合デシリアライザー）、その後プロデューサーを移行

詳細： 分類リファレンス

---

エッジケース

• モノレポ： Kafka 依存関係を持つ各サービス／モジュールを個別のアプリとして扱います
• 複数トピックを扱うプロデューサー： トピックごとにスキーマリソースを 1 つ生成します
• 共有スキーマ： スキーマファイルは 1 つ作成し、複数の Terraform リソースから参照します
• トピック名がない場合： 環境変数から読み込む場合は、TODO を含むプレースホルダーを使用します
• テストコード： スキーマ定義のみを含む場合を除き、テストディレクトリはスキップします
• 複数のシリアライザー： 形式ごとに個別のスキーマファイルを作成します

---

出力の構成

{project_root}/
├── schema-report.md              # Analysis report
├── schemas/
│   ├── schema.yaml               # Schema project config
│   ├── avro/
│   │   └── {topic}-value.avsc
│   ├── json/
│   │   └── {topic}-value.json
│   └── proto/
│       └── {topic}-value.proto
└── terraform/
    ├── providers.tf
    ├── variables.tf
    ├── tags.tf                    # PII/PRIVATE/SENSITIVE tags
    ├── schemas.tf                 # Active schemas (depends_on tags)
    ├── flagged-auto-register.tf   # Commented-out Category C
    ├── outputs.tf
    └── import.sh                  # Import existing schemas

---

リファレンスドキュメント

• 検出パターン — Kafka アプリ、依存関係、プロデューサー、コンシューマー、シリアライザーを見つけるためのパターン
• スキーマ推論 — データモデルやインラインデータからスキーマを抽出し、PII タグを付与します
• 分類 — カテゴリーの定義、ロールアウト順序、クライアントのバージョン要件
• Terraform テンプレート — ファイル構造、テンプレート、命名規則
• レポートテンプレート — 必須セクション、書式ルール、検証チェックリスト
• コード移行 — Python、Java、JavaScript、Go、.NET 向けのシリアライザー／デシリアライザー実装パターン

---

実行手順

1. Glob を使ってビルドファイルとスキーマファイルを見つけます
2. Grep を使ってパターン（依存関係、プロデューサー、シリアライザー、リスク）を検出します
3. Read を使ってソースファイルとデータモデルを調べます
4. Write を使ってスキーマファイル、Terraform 設定、レポートを作成します

Agent ツールを使う必要はありません — このスキルは自己完結しており、ツールを直接呼び出して使用します。