Governing Kafka Contracts: The Schema Registry Triad

Streaming estates rarely standardize on one serialization format. A real Schema Registry holds Avro subjects next to JSON Schema subjects next to Protobuf subjects, and every one of them is a contract some consumer depends on at runtime — which is exactly where drift surfaces: not in a code review, but in a deserialization exception at 2 a.m. Here is how to govern that estate from one place with CoreModels — including an honest account of where today's parsing boundary sits — and how the same .proto or .avsc becomes SQL DDL, JSON Schema, or an ODCS contract for the rest of your platform.

The triad problem

Confluent Schema Registry does its job well: it versions subjects and enforces compatibility rules per subject. What it does not do is tell you whether the order_status enum in your Avro subject still means the same thing as the order_status column in your warehouse, or whether a producer team quietly widened an enum that three consumer teams treat as closed. Compatibility checking is syntactic and per-subject; governance is semantic and cross-estate.

CoreModels treats the registry as one more estate to govern. The Confluent connector (vendor key confluent) has three capabilities: Import, Audit, and Generate. It never connects to your registry and never holds credentials — you upload the JSON the registry API already returns, and CoreModels works from that. Import writes additively to the governed graph; audit and generate never write anything.

Step 1: Export the subjects

The connector takes a single artifact named subjects: a JSON array of the registry's GET /subjects/{subject}/versions/latest responses. Aggregating it is a one-liner against your registry:

# Export every subject's latest schema from the Schema Registry:
for S in $(curl -s $REGISTRY/subjects | jq -r '.[]'); do
  curl -s "$REGISTRY/subjects/$S/versions/latest"
done | jq -s '.' > subjects.json

Each row is {subject, version, id, schemaType, schema} — and schemaType is where the triad lives. Avro, JSON Schema, and Protobuf subjects all ride in the same array; you do not sort them into buckets first.

Step 2: Import — Avro and JSON Schema parsed, Protobuf counted

Import is a single POST. Replace coremodels.example.com with your CoreModels host throughout; the interactive surface uses your normal login token:

curl -X POST "https://coremodels.example.com/graph/integrations/confluent/import/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"artifacts\": {\"subjects\": $(jq -Rs . < subjects.json)}}"

(The artifacts map takes raw content as strings, so the file is embedded with jq -Rs.)

CoreModels parses each subject by its declared schemaType, and is explicit about how far each format goes. Avro record schemas — the registry default — parse fully: union-null optionality (a non-nullable Avro field decodes as required), enums become accepted values, logical date/timestamp types are honored, and named-record references are resolved against the other subjects. JSON Schema subjects parse at depth 1 (properties, required, enum). Protobuf subjects are counted, not parsed, today — each one is reported in the import's lossiness ledger rather than silently skipped. Parsed datasets become governed types, their fields become governed elements with types, and enumerations become governed taxonomies, so a value set defined in an Avro subject and one defined in a JSON Schema subject are governed by the same machinery.

Two honesty notes, because they matter in production:

• Import is additive. Re-importing never mutates governed nodes you have already curated; changing governed meaning stays a human act.
• The Protobuf boundary is declared, not discovered. Every import files a lossiness record per Protobuf subject ("counted, not imported"), and every audit surfaces the count as a protobuf-unparsed Info finding in the Coverage section — so you always know exactly what is and is not under governance.

Import also stores a snapshot of the estate for later re-auditing. If the encoded snapshot exceeds the storage cap (around 1.5 MB), the import reports snapshotStored: false with a lossiness record; fresh-artifact audits still work, but reaudit has no stored snapshot to run against.

Step 3: Audit — catch drift before your consumers do

Audit compares a fresh subjects.json against the governed model. It is strictly read-only (Viewer role suffices) and by default records nothing:

curl -X POST "https://coremodels.example.com/graph/integrations/confluent/audit/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"artifacts\": {\"subjects\": $(jq -Rs . < subjects.json)}, \"recordHistory\": false}"

The report has three sections — Coverage, Drift, Conformance — with severities Error, Warning, Info:

{
  "success": true,
  "vendor": "confluent",
  "errorCount": 1,
  "warningCount": 0,
  "infoCount": 2,
  "codes": { "field-type-drift": 1, "fields-no-doc": 2 },
  "driftedObjects": ["..."],
  "fingerprint": "...",
  "findings": [
    {
      "section": "Drift",
      "severity": "Error",
      "code": "field-type-drift",
      "subject": "...",
      "message": "Field type drifted from the governed model.",
      "detail": "governed: ..., artifact: ..."
    }
  ],
  "markdown": "...",
  "historyRecorded": false,
  "lossiness": []
}

The finding codes are the real vocabulary of streaming drift:

The enum family deserves emphasis for Kafka work. Registry compatibility modes will happily accept many enum changes; whether your consumers can accept them is a semantic question, and enum-narrowed versus enum-widened versus enum-constraint-removed tells you precisely which kind of breakage you are looking at.

Step 4: Generate registry-ready Avro from governed entities

Governance is not just inbound. From the governed graph, CoreModels generates one schemas/{RecordName}.avsc per governed type:

curl -X POST "https://coremodels.example.com/graph/integrations/confluent/generate/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"typeNames": [], "extra": {}}'

(typeNames restricts generation to specific governed types; empty means everything eligible.) The emitted schemas are registry-ready Avro records: required fields appear bare, optional fields as ["null", T] with "default": null, governed taxonomies become sanitized Avro enums, and DateTime fields emit as timestamp-millis. Generate never writes to the graph — you take the .avsc files to your registry through whatever review process you already trust.

The transit: one contract, many dialects

Here is the part streaming teams usually do by hand: the analytics team wants the event as a warehouse table, the API team wants JSON Schema, the data-product owner wants an ODCS contract. The schema transformation engine runs stateless, schema-to-schema — the format keys avro and protobuf (alias proto) decode, and sql, jsonschema, and odcs are among the encode targets. And notably, this engine parses proto3 in full — the .proto file itself, not the registry subject, is the input here.

Over MCP, that is the transform_schema tool. Feed it a proto3 file straight from your registry export; the default inferred mapping aligns against a target hint, and for a pure format transit the schema itself is the hint:

{
  "tool": "transform_schema",
  "arguments": {
    "graphProjectId": "<project-id>",
    "sourceFormat": "protobuf",
    "sourceSchema": "syntax = \"proto3\";\n\nmessage Customer {\n  int64 customer_key = 1;\n  string full_name = 2;\n  optional string signup_date = 3;\n}",
    "targetFormat": "sql",
    "vendor": "postgres",
    "targetHintFormat": "protobuf",
    "targetHintSchema": "syntax = \"proto3\";\n\nmessage Customer {\n  int64 customer_key = 1;\n  string full_name = 2;\n  optional string signup_date = 3;\n}"
  }
}

Swap targetFormat for jsonschema or odcs (a Bitol Open Data Contract Standard v3 YAML contract), or set sourceFormat to avro for an .avsc source. The decode side follows the proto3 language spec: messages map to types (nested messages become child types), fields to elements, enums to taxonomies, repeated to collection cardinality, and google.protobuf.Timestamp to DateTime; plain scalar fields — which in proto3 always carry a value — decode as required, while explicit optional fields decode as optional. Every call returns the transformed schema plus a replayable plan and a lossiness ledger:

{
  "success": true,
  "schema": "...",
  "plan": { "operations": [ /* the replayable mapping plan */ ] },
  "lossiness": [
    { "kind": "ConstraintRelaxation", "path": "Element[...]", "explanation": "why" }
  ]
}

The ledger uses four kinds — StructuralDrop, TypeApproximation, ConstraintRelaxation, SemanticNarrowing — and it is the honest core of the product. Proto3 has no date-time scalar, so a DateTime degrading to string is recorded, not hidden. map<K,V> fields approximate as strings — recorded. Postgres has no inline enum, so an enum constraint relaxing to VARCHAR is recorded too. In our vocabulary, success: true does not mean "nothing changed"; it means "it ran." Always read lossiness.

The same transit is available over HTTP at POST graph/transform/schema/map/{projectId}, and once an estate is imported you can export the governed model itself — POST graph/transform/schema/export/{projectId} with { "format": "odcs" } hands the platform a data contract for what the streams mean, not just what they serialize.

Agents watching stream contracts

All of the read-side machinery is exposed over MCP at the OAuth-protected /mcp endpoint, so an agent can watch your streaming contracts without any standing credentials to your registry:

claude mcp add --transport http coremodels https://go.coremodels.io/mcp

The /mcp endpoint serves read-only tools; an agent session typically runs get_vendor_integration_status to see the connectors and the project's last-import state, then audit_vendor_project:

{
  "tool": "audit_vendor_project",
  "arguments": {
    "graphProjectId": "<project-id>",
    "vendor": "confluent",
    "artifactUrls": {
      "subjects": "https://ci.example.com/artifacts/subjects.json"
    }
  }
}

For registry exports too large to inline, artifactUrls lets the server fetch the artifact itself — https only, redirects disabled, private address ranges refused, response size capped. The tool is read-only and returns the same coverage/drift/conformance report; errorCount > 0 is the fail signal an agent should treat as "stop and tell a human." Writes — importing an estate with import_vendor_project, for instance — live on a separate admin endpoint and require write-level project membership (Admin, for the import), so a read-scoped watcher agent physically cannot mutate governed meaning.

Closing the loop: history, badge, CI

Drift is not an event, it is a pressure — so the audit belongs in the pipeline that changes contracts. On the machine surface (v1), user API keys work and the report rides in an ApiResponse envelope, so counts live under data.*:

RESPONSE=$(curl -s -X POST \
  "https://coremodels.example.com/v1/$PROJECT_ID/integrations/confluent/audit" \
  -H "Authorization: Bearer $COREMODELS_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"artifacts\": {\"subjects\": $(jq -Rs . < subjects.json)}, \"recordHistory\": true}")

ERRORS=$(echo "$RESPONSE" | jq '.data.errorCount')
if [ "$ERRORS" -gt 0 ]; then
  echo "Streaming contracts violate governed meaning ($ERRORS errors)."
  exit 1
fi

data.errorCount > 0 means the change violates governed meaning: fail the build. With recordHistory: true, each CI run is recorded in the audit history, which you can read back as a drift-over-time trail:

curl -H "Authorization: Bearer $TOKEN" \
  "https://coremodels.example.com/graph/integrations/confluent/history/$PROJECT_ID"

The latest recorded run also drives a shields-style SVG badge — green when clean, yellow for warnings only, red for errors, gray when nothing has been recorded. It is served on both surfaces; the v1 variant works with a user API key, which makes it embeddable in a README:

curl -H "Authorization: Bearer $COREMODELS_API_KEY" \
  "https://coremodels.example.com/v1/$PROJECT_ID/integrations/confluent/badge" > streaming-contracts.svg

And drift runs in both directions. When the governed model changes — someone tightens a taxonomy, retypes a field — you do not need a fresh registry export to know which streams are now out of line. reaudit replays the estate snapshot stored at import time against the current governed model, and always records its run:

curl -X POST "https://coremodels.example.com/graph/integrations/confluent/reaudit/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"projectName": null, "spaces": null}'

That is the full loop for a streaming estate: export subjects with one shell loop, import the triad additively, audit every contract change in CI with real finding codes, generate registry-ready Avro from governed meaning, and transit any .proto or .avsc into the dialects the rest of your platform speaks — with every approximation written down in a lossiness ledger instead of papered over. The registry keeps your schemas compatible; CoreModels keeps them meaning what you agreed they mean.

Get started with the Confluent quickstart in the CoreModels docs.