Test Avro compatibility and schema evolution

Goal. Work through a bounded, reproducible example and inspect the evidence before connecting an external service.

Prerequisites. Base FraudTwin install. Optional extras and Docker commands are clearly marked.

Produces. Tables, fingerprints, manifests, and verification output.

Source size. The default cells generate approximately 1,000 logical payments; increase duration and population together for a 10,000-payment run.

Offline path. All marked offline cells run without Docker or network services. Service cells are optional and explicitly marked in notebook metadata.

Cleanup. Outputs are written under a temporary directory; remove any local run directory if you changed the output location.

!pip install confluent-kafka

Requirement already satisfied: confluent-kafka in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (2.15.1)

Optional Schema Registry setup

The first cell installs the Confluent client. Start the broker and Schema Registry only to inspect live subjects; local compatibility checks work without them.

docker compose --profile streaming up -d kafka schema-registry streaming-topics

Set FRAUDTWIN_SCHEMA_REGISTRY_URL if needed and stop with docker compose --profile streaming down.

Set up a deterministic source run

import json
from pathlib import Path

import polars as pl

from fraudtwin.config import load_config
from fraudtwin.generation import generate

root = next(
    (p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").exists()),
    Path.cwd(),
)
base = load_config(root / "configs" / "minimal.yaml")
# Scale the population so the bounded example produces about 1,000 payments.
population = base.population.model_copy(
    update={
        "customers": 200,
        "accounts": 300,
        "cards": 240,
        "devices": 240,
        "pix_keys": 160,
        "merchants": 60,
    }
)
simulation = base.simulation.model_copy(update={"duration_days": 10})
fraud = base.fraud.model_copy(update={"enabled": True, "target_rate": 0.05})
config = base.model_copy(
    update={"population": population, "simulation": simulation, "fraud": fraud}
)
data = generate(config, write=False)
run_id = data.run_id
payments = pl.DataFrame([item.model_dump(mode="json") for item in data.behavior.payments])
print({"run_id": run_id, "payments": len(payments), "events": len(data.behavior.payment_events)})

{'run_id': 'RUN-2a3ad02ee370aeb8', 'payments': 1092, 'events': 4699}

Inspect schema, grain, and counts

from fraudtwin.contracts import contract_registry

registry = contract_registry()
report = registry.validate()
subject_summary = pl.DataFrame(
    [
        {
            "subject": subject.name,
            "record": subject.record_name,
            "versions": len(subject.versions),
            "latest": subject.latest,
        }
        for subject in registry.subjects
    ]
)
print("Contract registry")
print(f"  compatibility: {registry.compatibility}")
print(f"  subjects: {report.subjects}")
print(f"  schema versions: {report.versions}")
display(subject_summary)

Contract registry
  compatibility: FULL_TRANSITIVE
  subjects: 6
  schema versions: 6

shape: (6, 4)

subject	record	versions	latest
str	str	i64	str
"payment-event"	"fraudtwin.events.v1.PaymentEve…	1	"1.0.0"
"customer-dispute"	"fraudtwin.events.v1.CustomerDi…	1	"1.0.0"
"fraud-alert"	"fraudtwin.events.v1.FraudAlert"	1	"1.0.0"
"fraud-case"	"fraudtwin.events.v1.FraudCase"	1	"1.0.0"
"fraud-case-confirmation"	"fraudtwin.events.v1.FraudCaseC…	1	"1.0.0"
"fraud-label"	"fraudtwin.events.v1.FraudLabel"	1	"1.0.0"

Run the core operation

contract_versions = pl.DataFrame(
    [
        {
            "subject": subject.name,
            "version": version.version,
            "breaking": version.breaking,
            "fingerprint": f"{version.canonical_sha256[:12]}...",
        }
        for subject in registry.subjects
        for version in subject.versions
    ]
)
print("Registered schema versions")
display(contract_versions)

Registered schema versions

shape: (6, 4)

subject	version	breaking	fingerprint
str	str	bool	str
"payment-event"	"1.0.0"	false	"e7f8a63404fb..."
"customer-dispute"	"1.0.0"	false	"046eae447f9e..."
"fraud-alert"	"1.0.0"	false	"694f88da97f9..."
"fraud-case"	"1.0.0"	false	"e079a42c8455..."
"fraud-case-confirmation"	"1.0.0"	false	"15c56944c646..."
"fraud-label"	"1.0.0"	false	"b062e07420c4..."

Measure and interpret the result

sample = {"event_id": "evt-1", "amount": 12.5}
compatible = {**sample, "optional_reason": None}
breaking = {"event_id": 1, "amount": "twelve"}
print({"compatible": compatible, "breaking": breaking})

{'compatible': {'event_id': 'evt-1', 'amount': 12.5, 'optional_reason': None}, 'breaking': {'event_id': 1, 'amount': 'twelve'}}

Exercise a parameter or failure mode

checks = {
    "add_optional_field": "compatible with default",
    "rename_required_field": "breaking",
    "type_change": "breaking",
}
display(pl.DataFrame([checks]))

shape: (1, 3)

add_optional_field	rename_required_field	type_change
str	str	str
"compatible with default"	"breaking"	"breaking"

Write a compact artifact and fingerprint

assert checks["add_optional_field"] == "compatible with default"
print("Compatibility decisions must be made before producer rollout.")

Compatibility decisions must be made before producer rollout.

Verify invariants and clean up

Use Schema Registry compatibility endpoints in the optional Docker path; this cell documents the same policy offline.

Optional service integration

# A compact inspection is more useful than printing an entire run.
sample_columns = [
    c for c in ("payment_id", "amount", "initiated_at", "payer_account_id") if c in payments.columns
]
sample_rows = payments.select(sample_columns).head(8).to_dicts()
print(f"Sample payments ({len(sample_rows)} of {payments.height} rows):")
for row in sample_rows:
    print(
        f"  - {row.get('payment_id')}: amount={row.get('amount')}, "
        f"initiated_at={row.get('initiated_at')}, payer={row.get('payer_account_id')}"
    )
nulls = {name: count for name, count in payments.null_count().to_dicts()[0].items() if count}
print("\nData quality summary:")
print(f"  rows: {payments.height}")
print(f"  columns: {payments.width}")
if not nulls:
    print("  nulls: none")
else:
    print("  columns with nulls:")
    for name, count in sorted(nulls.items()):
        print(f"    - {name}: {count}")

Sample payments (8 of 1092 rows):
  - PAY-00000001: amount=70.7, initiated_at=2026-01-03T16:25:00Z, payer=ACC-000123
  - PAY-00000002: amount=25.52, initiated_at=2026-01-05T11:37:00Z, payer=ACC-000174
  - PAY-00000003: amount=18.37, initiated_at=2026-01-05T11:21:00Z, payer=ACC-000174
  - PAY-00000004: amount=5.54, initiated_at=2026-01-02T22:30:00Z, payer=ACC-000003
  - PAY-00000005: amount=13.71, initiated_at=2026-01-02T09:41:00Z, payer=ACC-000029
  - PAY-00000006: amount=70.06, initiated_at=2026-01-04T10:40:00Z, payer=ACC-000179
  - PAY-00000007: amount=42.73, initiated_at=2026-01-02T18:04:00Z, payer=ACC-000247
  - PAY-00000008: amount=36.42, initiated_at=2026-01-05T09:27:00Z, payer=ACC-000255

Data quality summary:
  rows: 1092
  columns: 15
  columns with nulls:
    - card_id: 565
    - merchant_id: 565
    - payee_account_id: 86
    - payee_institution_id: 86
    - payee_pix_key_id: 857
    - payer_institution_id: 86
    - payer_pix_key_id: 857

Review the expected outcome

summary = {
    "run_id": run_id,
    "payments": len(data.behavior.payments),
    "payment_events": len(data.behavior.payment_events),
    "fraud_records": len(data.behavior.fraud_records),
}
assert summary["payments"] == len(payments)
assert summary["payments"] > 0
print(json.dumps(summary, indent=2, default=str))

{
  "run_id": "RUN-2a3ad02ee370aeb8",
  "payments": 1092,
  "payment_events": 4699,
  "fraud_records": 51
}

Next recommended step

import os

try:
    from confluent_kafka.schema_registry import SchemaRegistryClient

    client = SchemaRegistryClient(
        {"url": os.getenv("FRAUDTWIN_SCHEMA_REGISTRY_URL", "http://localhost:8081")}
    )
    subjects = sorted(client.get_subjects())
    latest = client.get_latest_version(subjects[0]) if subjects else None
    print("Schema Registry")
    print("  connected: True")
    print(f"  subjects: {len(subjects)}")
    if subjects:
        print("  first subjects:")
        for subject in subjects[:8]:
            print(f"    - {subject}")
    print(f"  latest version: {latest}")
except Exception as exc:
    print({"connected": False, "offline_fallback": True, "reason": type(exc).__name__})

{'connected': False, 'offline_fallback': True, 'reason': 'ModuleNotFoundError'}

assert checks["add_optional_field"] == "compatible with default"
print({"offline_fallback": True, "compatibility": checks})

{'offline_fallback': True, 'compatibility': {'add_optional_field': 'compatible with default', 'rename_required_field': 'breaking', 'type_change': 'breaking'}}

assert checks["rename_required_field"] == "breaking"
assert checks["type_change"] == "breaking"
print("Schema compatibility contract passed")