kafka-schema-registry

Files (50)

• SKILL.md
• evals.json · evals/
• pom.xml · evals/mock-repos/acme-services/billing-service/
• Invoice.java · evals/mock-repos/acme-services/billing-service/src/main/java/com/acme/billing/
• requirements.txt · evals/mock-repos/acme-services/inventory-service/
• producer.py · evals/mock-repos/acme-services/inventory-service/src/
• consumer.py · evals/mock-repos/acme-services/reporting-service/src/
• pom.xml · evals/mock-repos/acme-services/user-service/
• UserEvent.java · evals/mock-repos/acme-services/user-service/src/main/java/com/acme/users/
• .env.example · evals/mock-repos/api-gateway/
• package.json · evals/mock-repos/api-gateway/
• middleware.js · evals/mock-repos/api-gateway/src/
• server.js · evals/mock-repos/api-gateway/src/
• requirements.txt · evals/mock-repos/ecommerce-platform/analytics-service/
• analytics_producer.py · evals/mock-repos/ecommerce-platform/analytics-service/src/
• package.json · evals/mock-repos/ecommerce-platform/notification-service/
• notification-producer.js · evals/mock-repos/ecommerce-platform/notification-service/src/
• pom.xml · evals/mock-repos/ecommerce-platform/order-service/
• OrderEvent.java · evals/mock-repos/ecommerce-platform/order-service/src/main/java/com/ecommerce/orders/
• OrderProducer.java · evals/mock-repos/ecommerce-platform/order-service/src/main/java/com/ecommerce/orders/
• go.mod · evals/mock-repos/go-kafka-producer/
• producer.go · evals/mock-repos/go-kafka-producer/
• requirements.txt · evals/mock-repos/legacy-event-publisher/
• event_publisher.py · evals/mock-repos/legacy-event-publisher/src/
• UserNotification.cs · evals/mock-repos/notification-service/Models/
• NotificationProducer.cs · evals/mock-repos/notification-service/
• NotificationService.csproj · evals/mock-repos/notification-service/
• config.yaml · evals/mock-repos/order-processor/
• requirements.txt · evals/mock-repos/order-processor/
• models.py · evals/mock-repos/order-processor/src/
• producer.py · evals/mock-repos/order-processor/src/
• pom.xml · evals/mock-repos/payment-service/
• PaymentEvent.java · evals/mock-repos/payment-service/src/main/java/com/example/payment/model/
• PaymentProducer.java · evals/mock-repos/payment-service/src/main/java/com/example/payment/producer/
• application.properties · evals/mock-repos/payment-service/src/main/resources/
• pom.xml · evals/mock-repos/streams-processor/
• StreamsApp.java · evals/mock-repos/streams-processor/src/main/java/com/example/streams/
• application.properties · evals/mock-repos/streams-processor/src/main/resources/
• pom.xml · evals/mock-repos/transaction-system/order-service/
• OrderCreatedEvent.java · evals/mock-repos/transaction-system/order-service/src/main/java/com/tx/orders/
• OrderProducer.java · evals/mock-repos/transaction-system/order-service/src/main/java/com/tx/orders/
• pom.xml · evals/mock-repos/transaction-system/payment-service/
• PaymentProcessedEvent.java · evals/mock-repos/transaction-system/payment-service/src/main/java/com/tx/payments/
• PaymentProducer.java · evals/mock-repos/transaction-system/payment-service/src/main/java/com/tx/payments/
• categorization.md · references/
• code-migration.md · references/
• detection-patterns.md · references/
• report-template.md · references/
• schema-inference.md · references/
• terraform-templates.md · references/

SKILL.md (first lines)

# Kafka Schema Registry Skill

Scan a project to identify Kafka applications, extract schemas, generate Terraform for Schema Registry registration, and produce a comprehensive analysis report.

## When to Use

Invoke this skill when:
- A user asks to analyze a project for Kafka usage in order to add event schemas or integrate Schema Registry
- A user wants to extract schemas from Kafka producers
- A user wants Terraform to register schemas to Schema Registry
- A user wants to audit Kafka producer/consumer configurations

## Deliverables

This skill produces 3 outputs in the target project:

1. **`schema-report.md`** — Full analysis report with findings, risks, and upgrade recommendations
2. **`schemas/`** — Extracted schema files (Avro, JSON Schema, Protobuf) with PII tagging
3. **`terraform/`** — Terraform configs using Confluent provider to register schemas

### Optional: Code Migration Assistance

If the user asks for their application code to be updated to integrate Schema Registry, use the [Code Migration Reference](references/code-migration.md) to update the code with proper Schema Registry integration patterns.

---

## High-Level Workflow

### Phase 0: Initialize

- Check for existing `schema.yaml` and `schemas/` directory manually
- Note any existing schema infrastructure in the report

### Phase 1: Project Scan & Kafka Detection

1. **Find build files** — Search for `pom.xml`, `build.gradle`, `requirements.txt`, `package.json`, etc.
2. **Detect Kafka dependencies** — Look for `spring-kafka`, `confluent-kafka`, `kafkajs`, etc.
3. **Find producers & consumers** — Grep for `KafkaTemplate`, `Producer(`, `producer.send`, etc.
4. **Extract topic names** — From string literals, config properties, YAML files
5. **Identify serializers** — Find `value.serializer`, `KafkaAvroSerializer`, custom serializers
6. **Build app catalog** — Compile findings: app name, language, role, topics, serializer, category

**Detailed patterns:** [Detection Patterns Reference](references/detection-patterns.md)

**App catalog structure:**
```yaml
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
```

**Multi-schema topic detection:**
- If multiple data models produce to the same topic, create a wrapper schema with `oneOf`/union/`oneof`
- Generate Terraform with `schema_reference` blocks
- Flag prominently in report

### Phase 2: Risk Detection

Search for:
- **`auto.register.schemas=true`** — Uncontrolled schema evolution (Category C)
- **`use.latest.version`** — Eases migration when set
- **Custom serializers** — Bypass SR entirely (Category E)

Record file path, line number, and affected topics for each occurrence.

**Patterns:** [Detection Patterns Reference](references/detection-patterns.md#risk-detection)

### Phase 3: Schema Inference

For each producer:
1. **Check for existing schema files** — `**/*.avsc`, `**/*.proto`, `**/*.schema.json`
2. **Infer from data models** — Java classes, Pydantic models, TypeScript interfaces, Go structs
3. **Infer from inline data** — HashMap, dict literals, map[string]any, plain objects, JSON strings
4. **Convert to schemas** — Map language types to JSON Schema / Avro / Protobuf
5. **Tag PII fields** — Scan field names for `email`, `ssn`, `phone`, `address`, etc.

**PII tagging:** Add `confluent:tags` (`PII`, `PRIVATE`, `SENSITIVE`, `PHI`) to detected fields.

**Detailed inference patterns:** [Schema Inference Reference](references/schema-inference.md)

### Phase 4: Categorize Producers

Classify each producer:

| Category | Criteria |
|----------|----------|
| **A: Compliant** | Confluent serializer + SR + no auto.register |
| **A→Header** | Already on SR, migrating to headers |
| **B: Schema in code, no SR** | Data models exist, but no SR integration |
| **C: Auto-register** | `auto.register.schemas=true` |
| **D: No schema** | Raw strings/bytes, no data model |
| **E: Custom serializer** | Custom `Serializer<T>` or inline serialization without SR |

**CRITICAL:** Use exact phrase "Category X" in:
- App catalog field
- Applications Discovered table
- Report section headers
- Terraform comments
- Risk sections

**Details:** [Categorization Reference](references/categorization.md)

### Phase 5: Create Schema Files

**Directory structure:**
```
schemas/
├── avro/
│   └── {topic}-value.avsc
├── json/
│   └── {topic}-value.json
└── proto/
    └── {topic}-value.proto

From confluentinc/agent-skills, Apache-2.0.

More from confluentinc/agent-skills