A payment service captures a card payment and records it in PostgreSQL. The rest of the business learns about it from an event on Kafka: a notifier emails the receipt, the ledger books the revenue. In the code, saving the row and publishing the event sit two lines apart, and in every test both happen.
In production, the process sometimes dies between those two lines: a deploy, an out-of-memory kill, a timeout. Commit first and die, and the payment is captured but no event leaves: no receipt, nothing in the ledger. Publish first and die before the commit, and every consumer hears about a capture the database rolled back. Neither shows up as an error, only later, when someone reconciles the ledger.
Earlier this year I fixed a version of this in a sync service between two SaaS systems. It created a record through one system's API, then wrote a mapping row in its own database; when the API call succeeded and the transaction rolled back, the retry created the record twice. Same shape: two writes to two places that can't commit together. That's the dual write problem.
You can't make a PostgreSQL commit and a Kafka produce atomic. The transactional outbox pattern stops trying: the service writes the event into an outbox table in the same transaction as the payment, and a separate relay publishes it afterwards. An event leaves if and only if the payment committed. The price is that the relay can publish an event more than once, so delivery becomes at-least-once, and the real work moves to duplicates and ordering. The cost is a table, a small relay process and a processed-events table in every consumer.
In this tutorial, you'll reproduce the dual write, then build an outbox in PostgreSQL, a polling relay and an idempotent Kafka consumer in Python, break each one, and finish with a chaos test of 1,000 payments. The API in front of this has its own retry problem, covered in How to Implement Idempotency Keys in Go with PostgreSQL. Everything was tested with Python 3.14.8, confluent-kafka 2.16.0, psycopg 3.3.6, PostgreSQL 18.6 and Apache Kafka 4.3.1.
Prerequisites
To follow along, you need:
- Docker with Docker Compose. The steps were tested on Docker 29.2.1. Python, PostgreSQL and Kafka all run in containers, so you don't install them locally.
- Working knowledge of Python and of SQL transactions.
- No Kafka experience is needed. Three terms carry the tutorial: a topic is a named log of messages split into partitions; a consumer group is a set of consumers that share a topic's partitions; and an offset is a consumer group's position in a partition, which it commits to Kafka as it goes.
- One terminal, open in an empty project folder.
Step 1: Set Up PostgreSQL and Kafka
First you need something to break: one PostgreSQL server with two databases, a single Kafka broker, and a Python image to run the scripts in. The payments database belongs to the payment service and notifier to the consumer that sends receipts; they share a server only to keep the lab small.
Create the project folder:
mkdir -p outbox-lab/sql && cd outbox-lab Create the Compose file:
compose.yaml
services: postgres: image: postgres:18.6 environment: POSTGRES_USER: app POSTGRES_PASSWORD: app POSTGRES_DB: payments volumes: - ./sql:/docker-entrypoint-initdb.d:ro healthcheck: test: ["CMD", "pg_isready", "-U", "app", "-d", "payments"] interval: 2s retries: 30 kafka: image: apache/kafka:4.3.1 environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" healthcheck: test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list"] interval: 5s retries: 30 app: build: . environment: PAYMENTS_DB: postgresql://app:app@postgres/payments NOTIFIER_DB: postgresql://app:app@postgres/notifier KAFKA: kafka:9092 volumes: - .:/app depends_on: postgres: condition: service_healthy kafka: condition: service_healthy profiles: ["tools"] The postgres service mounts sql/ into /docker-entrypoint-initdb.d, whose scripts the official image runs on its first start. Kafka runs in KRaft mode, acting as its own controller; Kafka 4.3 only supports KRaft, since ZooKeeper mode was removed. Topic auto-creation is off, so a mistyped topic name fails loudly. The app service sits in the tools profile, so docker compose up leaves it alone; you run each script with docker compose run.
The image only needs the two client libraries:
Dockerfile
FROM python:3.14.8-slim RUN pip install --no-cache-dir confluent-kafka==2.16.0 "psycopg[binary]==3.3.6" WORKDIR /app ENV PYTHONUNBUFFERED=1 The two SQL files create the schemas. Step 3 explains the outbox table and Step 6 the notifier's tables:
sql/01-payments.sql
CREATE TABLE payments ( id text PRIMARY KEY, amount_cents integer NOT NULL CHECK (amount_cents > 0), currency text NOT NULL, status text NOT NULL, updated_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE outbox ( id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, event_id uuid NOT NULL UNIQUE DEFAULT gen_random_uuid(), aggregate_id text NOT NULL, event_type text NOT NULL, payload jsonb NOT NULL, created_at timestamptz NOT NULL DEFAULT now(), published_at timestamptz ); CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL; sql/02-notifier.sql
CREATE DATABASE notifier; \connect notifier CREATE TABLE processed_events ( event_id uuid PRIMARY KEY, processed_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE receipts_sent ( id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, payment_id text NOT NULL, event_id uuid NOT NULL, sent_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE payment_status ( payment_id text PRIMARY KEY, status text NOT NULL ); common.py holds the connection settings and crash_point, which ends the process with os._exit(1), skipping finally blocks the way a SIGKILL would:
common.py
import os import random PAYMENTS_DB = os.environ["PAYMENTS_DB"] NOTIFIER_DB = os.environ["NOTIFIER_DB"] KAFKA = os.environ["KAFKA"] TOPIC = "payments" def crash_point(name: str) -> None: """Kill the process on the spot when CRASH_AT names this point, like a deploy or an OOM kill. CRASH_RATE makes it happen on only that fraction of passes.""" if os.environ.get("CRASH_AT") == name and random.random() < float(os.environ.get("CRASH_RATE", "1")): print(f"crash at {name}", flush=True) os._exit(1) These switches only stage failures
CRASH_AT and CRASH_RATE (in common.py), HOLD_SECONDS and SINGLE_ACTIVE (in relay.py) and DEDUPE (in notifier.py) exist only to stage the failures in this tutorial. Their defaults are the safe behavior; leave them unset in anything real.
Start PostgreSQL and Kafka, and build the Python image:
docker compose up -d postgres kafka docker compose build app Run docker compose ps until both services show (healthy), then create the topic:
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --partitions 3 --replication-factor 1 Kafka answers Created topic payments. The replication factor is 1 because there's only one broker.
By the end of Step 6, the pieces fit together like this:
View this diagram in the original post
The payment and its event commit together. The relay publishes the event later, and the notifier records each event ID it handles so a repeat does nothing.
Step 2: Reproduce the Dual-Write Problem
Most services start by writing to the database and publishing to Kafka in the same function, so that's the version to break first. naive.py does it in both possible orders:
naive.py
import json import sys import psycopg from confluent_kafka import Producer from common import KAFKA, PAYMENTS_DB, TOPIC, crash_point producer = Producer({"bootstrap.servers": KAFKA}) def publish(payment_id: str, event_type: str, data: dict) -> None: event = {"type": event_type, "payment_id": payment_id, "data": data} producer.produce(TOPIC, key=payment_id, value=json.dumps(event)) producer.flush(10) def capture_commit_first(conn: psycopg.Connection, payment_id: str) -> None: with conn.transaction(): conn.execute( "UPDATE payments SET status = 'captured', updated_at = now() WHERE id = %s", (payment_id,), ) crash_point("after_commit") publish(payment_id, "payment.captured", {}) def capture_publish_first(conn: psycopg.Connection, payment_id: str) -> None: with conn.transaction(): conn.execute( "UPDATE payments SET status = 'captured', updated_at = now() WHERE id = %s", (payment_id,), ) publish(payment_id, "payment.captured", {}) crash_point("before_commit") if __name__ == "__main__": order, payment_id = sys.argv[1], sys.argv[2] with psycopg.connect(PAYMENTS_DB) as conn: conn.execute( "INSERT INTO payments (id, amount_cents, currency, status) VALUES (%s, 2500, 'USD', 'authorized')", (payment_id,), ) conn.commit() if order == "commit-first": capture_commit_first(conn, payment_id) else: capture_publish_first(conn, payment_id) print(f"{payment_id} captured") capture_commit_first commits, then publishes. capture_publish_first publishes inside the open transaction, and producer.flush(10) waits for Kafka's acknowledgement, so the event is on the broker before the commit. Run each order with a crash staged between its two writes:
docker compose run --rm -e CRASH_AT=after_commit app python naive.py commit-first pay_001 Output
crash at after_commit Then run:
docker compose run --rm -e CRASH_AT=before_commit app python naive.py publish-first pay_002 Output
crash at before_commit Compare the database with the topic:
docker compose exec postgres psql -U app -d payments -c "SELECT id, status FROM payments ORDER BY id;" docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic payments --from-beginning --formatter-property print.key=true --timeout-ms 5000 Output
id | status ---------+------------ pay_001 | captured pay_002 | authorized pay_002 {"type": "payment.captured", "payment_id": "pay_002", "data": {}} Processed a total of 1 messages The console consumer also prints a hint about the new consumer protocol and a TimeoutException when --timeout-ms runs out; both are harmless. Look at the two payments side by side. pay_001 is captured in the database, but no event left, so no receipt goes out. pay_002 was never captured, yet consumers were told it was: that's a phantom event. A try/except can't fix either one, because a killed process runs no handler.
Why not Kafka transactions?
Kafka's transactions and idempotent producer make writes into Kafka exactly-once. They don't extend to your database. The Kafka design documentation says exactly-once delivery into other systems generally needs those systems' cooperation, with Kafka providing the primitives (it points to Kafka Connect), and that otherwise Kafka guarantees at-least-once delivery by default. A PostgreSQL commit can't join a Kafka transaction, so the gap between the two writes stays.
Before you move on, clear the payments tables and recreate the topic. The phantom event lacks the event_id that the consumer in Step 6 relies on:
docker compose exec postgres psql -U app -d payments -c "TRUNCATE payments, outbox RESTART IDENTITY;" docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic payments docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --partitions 3 --replication-factor 1 Step 3: Write Events with the Transactional Outbox Pattern
Both failures come from asking two systems to agree on one commit. The fix is to ask only one. The service inserts the event into the outbox table from Step 1, in the same transaction as the payment change, so the two commit together or not at all.
Each column of outbox has one job:
| Column | Job |
|---|---|
id | Identity column; the relay publishes in this order |
event_id | A UUID that consumers use to recognize a repeat |
aggregate_id | The payment ID; it becomes the Kafka message key |
event_type, payload | What happened, and its data |
created_at | When it happened; also the basis for the lag alert in Step 9 |
published_at | NULL until the relay has confirmed delivery |
The partial index outbox_unpublished covers only rows where published_at IS NULL, so the relay's query stays small however large the table grows, as long as autovacuum keeps up. Create the payment service:
payments.py
import sys import psycopg from psycopg.types.json import Jsonb from common import PAYMENTS_DB def add_event(conn: psycopg.Connection, payment_id: str, event_type: str, data: dict) -> None: conn.execute( "INSERT INTO outbox (aggregate_id, event_type, payload) VALUES (%s, %s, %s)", (payment_id, event_type, Jsonb(data)), ) def authorize(conn: psycopg.Connection, payment_id: str, amount_cents: int) -> None: with conn.transaction(): conn.execute( "INSERT INTO payments (id, amount_cents, currency, status) VALUES (%s, %s, 'USD', 'authorized')", (payment_id, amount_cents), ) add_event(conn, payment_id, "payment.authorized", {"amount_cents": amount_cents, "currency": "USD"}) def capture(conn: psycopg.Connection, payment_id: str) -> None: with conn.transaction(): row = conn.execute( """UPDATE payments SET status = 'captured', updated_at = now() WHERE id = %s AND status = 'authorized' RETURNING amount_cents, currency""", (payment_id,), ).fetchone() if row is None: raise ValueError(f"{payment_id} is not authorized") add_event(conn, payment_id, "payment.captured", {"amount_cents": row[0], "currency": row[1]}) if __name__ == "__main__": command, payment_id = sys.argv[1], sys.argv[2] with psycopg.connect(PAYMENTS_DB, autocommit=True) as conn: if command == "authorize": authorize(conn, payment_id, int(sys.argv[3])) else: capture(conn, payment_id) print(f"{payment_id}: {command} recorded") add_event runs inside the same conn.transaction() block as the payment write. In capture, the WHERE status = 'authorized' guard makes a second capture match no rows, and the ValueError rolls back the whole block, outbox insert included.
Authorize and capture one payment, and authorize a second:
docker compose run --rm app python payments.py authorize pay_001 2500 docker compose run --rm app python payments.py capture pay_001 docker compose run --rm app python payments.py authorize pay_002 4000 Each command prints a confirmation such as pay_001: authorize recorded. Check the outbox:
docker compose exec postgres psql -U app -d payments -c "SELECT id, aggregate_id, event_type, published_at FROM outbox ORDER BY id;" You'll see three rows, ids 1 to 3, each with an empty published_at. Nothing has reached Kafka yet, and that's fine: the events are safe in PostgreSQL.
Step 4: Publish the Outbox with a Relay
Those events are durable, but no consumer can see them. Getting them to Kafka is the relay's job: a separate process that loops: lock a batch of unpublished rows, produce them, wait for Kafka's acknowledgement, mark them published, commit. Create it:
relay.py
import json import os import time import psycopg from confluent_kafka import KafkaException, Producer from common import KAFKA, PAYMENTS_DB, TOPIC, crash_point BATCH_SIZE = 100 RELAY_LOCK = 4200 # any bigint; every relay instance must use the same one FETCH = """ SELECT id, event_id, aggregate_id, event_type, payload, created_at FROM outbox WHERE published_at IS NULL ORDER BY id LIMIT %s FOR UPDATE SKIP LOCKED """ def publish_batch(conn: psycopg.Connection, producer: Producer) -> int: with conn.transaction(): rows = conn.execute(FETCH, (BATCH_SIZE,)).fetchall() if not rows: return 0 time.sleep(float(os.environ.get("HOLD_SECONDS", "0"))) errors = [] def on_delivery(err, msg): if err is not None: errors.append(err) for _id, event_id, aggregate_id, event_type, payload, created_at in rows: event = { "event_id": str(event_id), "type": event_type, "payment_id": aggregate_id, "occurred_at": created_at.isoformat(), "data": payload, } producer.produce(TOPIC, key=aggr...
