Recover from Kafka outages and duplicate delivery#
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[schemaregistry]" httpx2
Collecting httpx2
Downloading httpx2-2.13.1-py3-none-any.whl.metadata (9.8 kB)
Requirement already satisfied: confluent-kafka[schemaregistry] in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (2.15.1)
Requirement already satisfied: attrs>=21.2.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from confluent-kafka[schemaregistry]) (26.1.0)
Requirement already satisfied: cachetools>=5.5.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from confluent-kafka[schemaregistry]) (7.2.0)
Requirement already satisfied: certifi in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from confluent-kafka[schemaregistry]) (2026.7.22)
Requirement already satisfied: httpx>=0.26 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from confluent-kafka[schemaregistry]) (0.28.1)
Requirement already satisfied: authlib>=1.0.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from confluent-kafka[schemaregistry]) (1.8.0)
Requirement already satisfied: anyio>=4.10 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from httpx2) (4.15.1)
Collecting httpcore2==2.13.1 (from httpx2)
Downloading httpcore2-2.13.1-py3-none-any.whl.metadata (26 kB)
Requirement already satisfied: idna>=3.18 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from httpx2) (3.19)
Collecting truststore>=0.10 (from httpx2)
Downloading truststore-0.10.4-py3-none-any.whl.metadata (4.4 kB)
Requirement already satisfied: h11>=0.16 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from httpcore2==2.13.1->httpx2) (0.16.0)
Requirement already satisfied: typing_extensions>=4.16.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from anyio>=4.10->httpx2) (4.16.0)
Requirement already satisfied: cryptography>=45.0.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from authlib>=1.0.0->confluent-kafka[schemaregistry]) (50.0.1)
Requirement already satisfied: joserfc>=1.6.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from authlib>=1.0.0->confluent-kafka[schemaregistry]) (1.7.5)
Requirement already satisfied: cffi>=2.0.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from cryptography>=45.0.1->authlib>=1.0.0->confluent-kafka[schemaregistry]) (2.1.1)
Requirement already satisfied: pycparser in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from cffi>=2.0.0->cryptography>=45.0.1->authlib>=1.0.0->confluent-kafka[schemaregistry]) (3.0)
Requirement already satisfied: httpcore==1.* in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from httpx>=0.26->confluent-kafka[schemaregistry]) (1.0.9)
Downloading httpx2-2.13.1-py3-none-any.whl (95 kB)
Downloading httpcore2-2.13.1-py3-none-any.whl (83 kB)
Downloading truststore-0.10.4-py3-none-any.whl (18 kB)
Installing collected packages: truststore, httpcore2, httpx2
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 3/3 [httpx2]━━━━━━━━━━━ 2/3 [httpx2]
Successfully installed httpcore2-2.13.1 httpx2-2.13.1 truststore-0.10.4
Note: you may need to restart the kernel to use updated packages.
Optional Kafka setup#
The first cell installs the Kafka client and Schema Registry support. Start Kafka and Schema Registry only for broker-backed recovery; outage injection and analysis also run offline.
Run the next Python cell from this notebook. It starts the required Docker services, finds the repository’s docker-compose.yml, and waits for Kafka, the payment topic, and Schema Registry. You do not need to paste a Docker command into a separate terminal.
The notebook uses localhost:9092 and http://localhost:8081 when it runs on the host. To stop the services later, run the cleanup cell near the end of the notebook. If the repository cannot be located automatically, set FRAUDTWIN_REPO_ROOT to the project directory before running the setup cell.
import os
import subprocess
import time
from pathlib import Path
from urllib.request import urlopen
repo_root = os.environ.get("FRAUDTWIN_REPO_ROOT")
candidates = ([Path(repo_root)] if repo_root else []) + [Path.cwd(), *Path.cwd().parents]
root = next(
(path.resolve() for path in candidates if (path / "docker-compose.yml").is_file()),
None,
)
if root is None:
raise RuntimeError(
"Could not find docker-compose.yml. Start Jupyter from the FraudTwin repository or "
"set FRAUDTWIN_REPO_ROOT to its path."
)
compose_env = os.environ.copy()
compose_env.setdefault("GRAFANA_ADMIN_PASSWORD", "fraudtwin-local")
compose_command = [
"docker",
"compose",
"--profile",
"streaming",
"up",
"-d",
"kafka",
"schema-registry",
"streaming-topics",
]
subprocess.run(
compose_command,
check=True,
cwd=root,
env=compose_env,
)
for _ in range(30):
try:
with urlopen("http://localhost:8081/subjects", timeout=2) as response:
if response.status == 200:
break
except OSError:
pass
time.sleep(2)
else:
raise RuntimeError("Schema Registry did not become ready; inspect docker compose logs")
expected_topic = "fraudsim.payment.events.v1"
for _ in range(30):
try:
topics = subprocess.run(
[
"docker",
"compose",
"--profile",
"streaming",
"exec",
"-T",
"kafka",
"kafka-topics",
"--bootstrap-server",
"kafka:29092",
"--list",
],
check=True,
capture_output=True,
text=True,
cwd=root,
env=compose_env,
).stdout.splitlines()
if expected_topic in topics:
break
except (OSError, subprocess.CalledProcessError):
pass
time.sleep(2)
else:
raise RuntimeError("Kafka did not become ready; inspect docker compose logs")
running_services = subprocess.run(
[
"docker",
"compose",
"--profile",
"streaming",
"ps",
"--services",
"--filter",
"status=running",
],
check=True,
capture_output=True,
text=True,
cwd=root,
env=compose_env,
).stdout.splitlines()
print(
{
"running_services": running_services,
"schema_registry": "ready",
"payment_topic": expected_topic,
}
)
Container fraudtwin-kafka-1 Running
Container fraudtwin-schema-registry-1 Running
Container fraudtwin-streaming-topics-1 Starting
Container fraudtwin-streaming-topics-1 Started
{'running_services': ['grafana', 'iceberg-rest', 'kafka', 'minio', 'prometheus', 'schema-registry', 'streaming-topics'], 'schema_registry': 'ready', 'payment_topic': 'fraudsim.payment.events.v1'}
The setup cell must report schema_registry: ready and payment_topic: fraudsim.payment.events.v1 before the live publication check. A connected: True result in the next integration cell confirms that records were published through the broker.
Set up a deterministic source run
import json
import os
from pathlib import Path
import polars as pl
from fraudtwin.config import load_config
from fraudtwin.generation import generate
repo_root = os.environ.get("FRAUDTWIN_REPO_ROOT")
candidates = ([Path(repo_root)] if repo_root else []) + [Path.cwd(), *Path.cwd().parents]
root = next(
(p.resolve() for p in candidates if (p / "configs" / "minimal.yaml").exists()),
None,
)
if root is None:
raise RuntimeError(
"Could not find configs/minimal.yaml. Set FRAUDTWIN_REPO_ROOT to the project path."
)
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.kafka import publication_records
from fraudtwin.kafka_chaos import KafkaChaosConfig, KafkaChaosOutage, simulate_delivery
records = publication_records(data.behavior, run_id)[:80]
print({"sent": len(records), "topics": sorted({r.topic for r in records})})
{'sent': 80, 'topics': ['fraudsim.payment.events.v1']}
Run the core operation
chaos_config = KafkaChaosConfig(
seed=11,
drop_probability=0.03,
duplicate_probability=0.08,
retry_probability=0.1,
max_delay_seconds=60,
reorder_window=10,
partition_count=3,
partition_skew_probability=0.2,
)
result = simulate_delivery(records, chaos_config)
print(result.manifest["counts"])
{'sent': 80, 'acknowledged': 89, 'dropped': 7, 'retried': 8, 'duplicated': 8, 'late': 84, 'reordered': 47, 'deduplicated': 16}
Measure and interpret the result
outage = KafkaChaosOutage(
from_time=records[0].observable_time,
to_time=records[-1].observable_time,
behavior="BUFFER_AND_FLUSH",
)
outage_result = simulate_delivery(records, chaos_config.model_copy(update={"outages": (outage,)}))
print("outage counts:", outage_result.manifest["counts"])
outage counts: {'sent': 80, 'acknowledged': 89, 'dropped': 7, 'retried': 8, 'duplicated': 8, 'late': 88, 'reordered': 47, 'deduplicated': 16}
Exercise a parameter or failure mode
deduped = {(e.subject, e.record_id): e for e in result.envelopes}
print({"delivered": len(result.envelopes), "deduplicated": len(deduped), "late": result.late_count})
{'delivered': 89, 'deduplicated': 73, 'late': 84}
Write a compact artifact and fingerprint
assert all(e.record_id for e in deduped.values())
print("stable event identity enables safe consumer deduplication")
stable event identity enables safe consumer deduplication
Verify invariants and clean up
display(pl.DataFrame(result.audit[:12]))
shape: (7, 2)
| record_id | fault |
|---|---|
| str | str |
| "EVT-HN-F03-000003-SIGNAL-01" | "drop" |
| "EVT-F01-000001-000001-02" | "drop" |
| "EVT-HN-F05-000005-000001-02" | "drop" |
| "EVT-HN-F05-000005-000002-02" | "drop" |
| "EVT-HN-F04-000004-000001-05" | "drop" |
| "EVT-HN-F05-000005-000004-02" | "drop" |
| "EVT-HN-F05-000005-000004-03" | "drop" |
Optional: publish the run to Kafka#
The offline cells above exercise deterministic loss, retry, duplication, delay, reordering, and deduplication without a broker. Run the live publication cell only after the setup cell reports that Kafka, Schema Registry, and the payment topic are ready. A successful broker-backed check prints connected: True and publication counts; otherwise the cell reports the reason and the offline path remains valid.
Watermarks must account for delayed/out-of-order event time; physical packet loss is outside this logical harness. The outage simulation models the delivery semantics that consumers must handle after transport recovers.
Inspect the generated run#
# 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
Confirm the generated run#
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
}
Publish to a live Kafka broker (optional)#
import os
os.environ.setdefault("FRAUDTWIN_KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
os.environ.setdefault("FRAUDTWIN_SCHEMA_REGISTRY_URL", "http://localhost:8081")
try:
from confluent_kafka.admin import AdminClient
from fraudtwin.kafka import publisher_from_environment
admin = AdminClient(
{"bootstrap.servers": os.getenv("FRAUDTWIN_KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")}
)
topics = sorted(admin.list_topics(timeout=3).topics)
published = publisher_from_environment().publish(data.behavior, run_id)
print({"connected": True, "topics": topics, "published": published.record_counts})
except Exception as exc:
print(
{
"connected": False,
"offline_fallback": True,
"reason": type(exc).__name__,
"detail": str(exc),
}
)
{'connected': True, 'topics': ['__consumer_offsets', '_schemas', 'fraudsim.customer.disputes.v1', 'fraudsim.fraud.alerts.v1', 'fraudsim.fraud.case-confirmations.v1', 'fraudsim.fraud.cases.v1', 'fraudsim.fraud.labels.v1', 'fraudsim.payment.events.v1'], 'published': {'payment-event': 4699, 'customer-dispute': 46, 'fraud-alert': 51, 'fraud-case': 51, 'fraud-case-confirmation': 51, 'fraud-label': 51}}
Compare with offline recovery#
This final cell records the deterministic recovery result from the logical outage simulation. It does not require Kafka and remains reproducible when the optional services are stopped.
assert result.output_fingerprint
print({"offline_fallback": True, "recovered": len(deduped)})
{'offline_fallback': True, 'recovered': 73}