kafka-schema-registry

Kafka Schema Registry 技能

扫描项目以识别 Kafka 应用、提取架构、生成用于向 Schema Registry 注册的 Terraform，并生成全面的分析报告。

使用场景

在以下情况下调用此技能：

• 用户要求分析项目中的 Kafka 用法，以便添加事件架构或集成 Schema Registry
• 用户想从 Kafka 生产者中提取架构
• 用户想使用 Terraform 向 Schema Registry 注册架构
• 用户想审计 Kafka 生产者/消费者配置

交付内容

此技能会在目标项目中生成 3 项输出：

1. schema-report.md — 完整的分析报告，包含发现、风险和升级建议
2. schemas/ — 提取的架构文件（Avro、JSON Schema、Protobuf），并附有 PII 标记
3. terraform/ — 使用 Confluent provider 注册架构的 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 等
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，正在迁移到 headers
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

文件命名：必须使用烤肉串命名法（小写字母，以连字符分隔）：

• 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 列为必需项）
• 风险（自动注册、自定义序列化器）
• 生产者升级建议（针对每个应用，标题中包含「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 的服务/模块视为独立应用
• **多主题生产者：**每个主题生成一个架构资源
• **共享架构：**创建一个架构文件，由多个 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 工具 — 此技能是独立的，直接调用工具即可。