How to Implement the Transactional Outbox Pattern in Python

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=aggregate_id, value=json.dumps(event), on_delivery=on_delivery)

        if producer.flush(30) > 0 or errors:
            raise KafkaException(errors[0] if errors else "delivery timed out")
        crash_point("after_produce")

        conn.execute(
            "UPDATE outbox SET published_at = now() WHERE id = ANY(%s)",
            ([r[0] for r in rows],),
        )
    print(f"published {len(rows)}: {', '.join(f'{r[3]} {r[2]}' for r in rows[:5])}", flush=True)
    return len(rows)


def main() -> None:
    producer = Producer({"bootstrap.servers": KAFKA, "enable.idempotence": True})
    with psycopg.connect(PAYMENTS_DB, autocommit=True) as conn:
        if os.environ.get("SINGLE_ACTIVE", "1") == "1":
            while not conn.execute("SELECT pg_try_advisory_lock(%s)", (RELAY_LOCK,)).fetchone()[0]:
                print("standby: another relay is active", flush=True)
                time.sleep(2)
            print("active", flush=True)
        while True:
            if publish_batch(conn, producer) == 0:
                time.sleep(0.5)


if __name__ == "__main__":
    main()

FETCH takes the 100 oldest unpublished rows and locks them with FOR UPDATE SKIP LOCKED, so a second relay skips rows the first one holds instead of waiting, a mode the PostgreSQL documentation reserves for queue-like tables because it gives an inconsistent view of the data.

Each event is produced with the payment ID as its key. The producer's partitioner hashes the key to pick a partition, so every event for one payment lands in the same partition, and Kafka's introduction promises that "any consumer of a given topic-partition will always read that partition's events in exactly the same order as they were written."

In confluent-kafka-python, produce() only queues the message, and the delivery callback runs later, from poll() or flush(). flush(30) returns the number of messages still queued, so the relay treats anything above zero, or any error passed to on_delivery, as a failure. The exception rolls the transaction back, leaving the rows unpublished, and stops the relay, so run it under a supervisor (systemd, a container restart policy) that starts it again. Only after Kafka has acknowledged every message does the UPDATE set published_at.

The producer sets enable.idempotence, which defaults to false in librdkafka, the C library under the Python client. It implies acks=all, and the same reference warns that "retrying may cause reordering unless enable.idempotence is set to true."

There's a trade-off here. The relay holds its transaction and row locks open while it waits for Kafka, up to 30 seconds per batch on a struggling broker. I'd rather accept that than mark rows before Kafka confirms them, which turns a slow broker into lost events.

Why the Relay Doesn't Track the Last ID It Sent

You might reach for a simpler cursor instead: remember the last id you sent and poll WHERE id > last_seen. This script shows how that loses events:

cursor_gap.py

"""Why a relay must not poll with WHERE id > last_seen_id."""
import threading
import time

import psycopg
from psycopg.types.json import Jsonb

from common import PAYMENTS_DB

INSERT = "INSERT INTO outbox (aggregate_id, event_type, payload) VALUES (%s, 'payment.authorized', %s) RETURNING id"


def slow_transaction():
    with psycopg.connect(PAYMENTS_DB) as conn:
        event_id = conn.execute(INSERT, ("pay_slow", Jsonb({}))).fetchone()[0]
        print(f"slow transaction took id {event_id}, commits in 2s")
        time.sleep(2)
    print(f"slow transaction committed id {event_id}")


with psycopg.connect(PAYMENTS_DB, autocommit=True) as conn:
    last_seen = conn.execute("SELECT coalesce(max(id), 0) FROM outbox").fetchone()[0]
    slow = threading.Thread(target=slow_transaction)
    slow.start()
    time.sleep(0.5)
    fast_id = conn.execute(INSERT, ("pay_fast", Jsonb({}))).fetchone()[0]
    print(f"fast transaction committed id {fast_id}")

    seen = [r[0] for r in conn.execute("SELECT id FROM outbox WHERE id > %s ORDER BY id", (last_seen,))]
    print(f"cursor relay reads ids {seen}, moves last_seen to {max(seen)}")
    last_seen = max(seen)
    slow.join()
    later = [r[0] for r in conn.execute("SELECT id FROM outbox WHERE id > %s ORDER BY id", (last_seen,))]
    print(f"next poll reads ids {later}: id {fast_id - 1} is never published")
    conn.execute("DELETE FROM outbox WHERE aggregate_id IN ('pay_slow', 'pay_fast')")

Then run:

docker compose run --rm app python cursor_gap.py

Output

slow transaction took id 4, commits in 2s
fast transaction committed id 5
cursor relay reads ids [5], moves last_seen to 5
slow transaction committed id 4
next poll reads ids []: id 4 is never published

Look at the last line: id 4 committed without error, and the cursor will never see it. An identity value is assigned at insert, but the row becomes visible only at commit, and transactions don't commit in the order they started (sequences aren't gapless either). The cursor moves past the fast transaction's id 5, and id 4 is skipped forever. Polling published_at IS NULL has no cursor, so a late commit is picked up by the next poll. The script deletes its two rows when it finishes.

Hold off on starting the relay. In Step 5 you'll start it set up to fail.

Step 5: Crash the Relay and Watch Events Duplicate

Marking rows only after Kafka confirms them leaves one window open: Kafka has acknowledged a batch, but the UPDATE marking it published hasn't committed. A crash there looks like this:

View this diagram in the original post

Kafka has the batch, PostgreSQL doesn't know it. The restarted relay can only send it again.

Run the relay with a crash staged at exactly that point:

docker compose run --rm -e CRASH_AT=after_produce app python relay.py

Output

active
crash at after_produce

The relay produced all three events and died before it could record that. A GETPID warning may also appear on a fresh broker (see Troubleshooting). Confirm that nothing was marked:

docker compose exec postgres psql -U app -d payments -c "SELECT count(*) AS unpublished FROM outbox WHERE published_at IS NULL;"

The count is still 3: as far as PostgreSQL knows, nothing went out. Now start the relay for real, in the background:

docker compose run -d --name relay app python relay.py

Wait about 10 seconds, then read its log:

docker logs relay

Output

active
published 3: payment.authorized pay_001, payment.captured pay_001, payment.authorized pay_002

It sent all three events again. Count what reached the topic:

docker compose exec kafka /opt/kafka/bin/kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic payments

Output

payments:0:2
payments:1:4
payments:2:0

Partition 0 ends at offset 2 and partition 1 at offset 4: six messages for three events. Every event is on the topic twice, with the same event_id. Leave the relay running.

You might expect enable.idempotence to catch this. It doesn't, and it can't: the broker deduplicates retries by a producer ID and sequence number, and the restarted relay is a new producer. KIP-98, the proposal that introduced the idempotent producer, states the limit:

Since each new instance of a producer is assigned a new, unique, PID, we can only guarantee idempotent production within a single producer session.

A hand-written relay that marks rows in PostgreSQL can't close this window, only make it rarer. Kafka Connect's exactly-once source mode attempts to close it for connectors such as Debezium, and consumers still see repeats when they crash before committing an offset. The pattern's own description warns that the message relay might publish a message more than once. So the outbox gives you at-least-once delivery , and the consumer has to cope.

Step 6: Make the Consumer Idempotent

If the relay can send an event twice, the consumer is where duplicates have to stop. The notifier turns payment.captured into a receipt and keeps a payment_status projection, its own copy of each payment's status. To survive repeats it uses an inbox table, processed_events, keyed by event_id:

notifier.py

import json
import os
import signal
import sys

import psycopg
from confluent_kafka import Consumer, KafkaException

from common import KAFKA, NOTIFIER_DB, TOPIC, crash_point

DEDUPE = os.environ.get("DEDUPE", "1") == "1"


def handle(conn: psycopg.Connection, event: dict) -> str:
    with conn.transaction():
        if DEDUPE:
            claimed = conn.execute(
                "INSERT INTO processed_events (event_id) VALUES (%s) ON CONFLICT DO NOTHING",
                (event["event_id"],),
            ).rowcount
            if not claimed:
                return "duplicate, skipped"
        status = event["type"].removeprefix("payment.")
        conn.execute(
            """INSERT INTO payment_status (payment_id, status) VALUES (%s, %s)
               ON CONFLICT (payment_id) DO UPDATE SET status = EXCLUDED.status""",
            (event["payment_id"], status),
        )
        if event["type"] == "payment.captured":
            conn.execute(
                "INSERT INTO receipts_sent (payment_id, event_id) VALUES (%s, %s)",
                (event["payment_id"], event["event_id"]),
            )
            return "receipt sent"
        return "status updated"


def main() -> None:
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    consumer = Consumer({
        "bootstrap.servers": KAFKA,
        "group.id": "notifier",
        "enable.auto.commit": False,
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe([TOPIC])
    with psycopg.connect(NOTIFIER_DB, autocommit=True) as conn:
        try:
            while True:
                msg = consumer.poll(1.0)
                if msg is None:
                    continue
                if msg.error():
                    raise KafkaException(msg.error())
                event = json.loads(msg.value())
                outcome = handle(conn, event)
                crash_point("before_offset_commit")
                consumer.commit(message=msg, asynchronous=False)
                print(f"p{msg.partition()}@{msg.offset()} {event['type']} {event['payment_id']}: {outcome}", flush=True)
        finally:
            consumer.close()


if __name__ == "__main__":
    main()

handle runs in one transaction. It claims the event with INSERT ... ON CONFLICT DO NOTHING, and rowcount says whether the event_id was new. A new event updates the projection and, for a capture, records the receipt; a repeat returns early. The claim and the effects commit together. This is the consumer-side twin of the key table in the idempotency keys tutorial. receipts_sent stands in for the email; a real email provider sits outside this transaction, so pass it the event_id as its own idempotency key.

The offset is committed only after the database commit, with enable.auto.commit off and commit(message=msg, asynchronous=False). A committed offset points at the next message to read, so a crash before the commit means the event comes back, and the inbox turns it away.

First, see what happens without the inbox. Start the notifier with DEDUPE=0:

docker compose run -d --name notifier -e DEDUPE=0 app python notifier.py

Wait about 10 seconds, then read its log:

docker logs notifier

Output

p1@0 payment.authorized pay_001: status updated
p1@1 payment.captured pay_001: receipt sent
p1@2 payment.authorized pay_001: status updated
p1@3 payment.captured pay_001: receipt sent
p0@0 payment.authorized pay_002: status updated
p0@1 payment.authorized pay_002: status updated

Each line is partition, offset, event and outcome. Notice receipt sent twice for pay_001. Count the receipts:

docker compose exec postgres psql -U app -d notifier -c "SELECT payment_id, count(*) AS receipts FROM receipts_sent GROUP BY payment_id;"

pay_001 has 2 receipts for one payment. Stop the notifier, empty its tables and rewind the group:

docker stop notifier && docker rm notifier
docker compose exec postgres psql -U app -d notifier -c "TRUNCATE processed_events, receipts_sent, payment_status;"
docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group notifier --reset-offsets --to-earliest --topic payments --execute

The reset prints a new offset of 0 for each partition. Now run it with the inbox on, the default:

docker compose run -d --name notifier app python notifier.py

Wait about 10 seconds, then read its log:

docker logs notifier

Output

p0@0 payment.authorized pay_002: status updated
p0@1 payment.authorized pay_002: duplicate, skipped
p1@0 payment.authorized pay_001: status updated
p1@1 payment.captured pay_001: receipt sent
p1@2 payment.authorized pay_001: duplicate, skipped
p1@3 payment.captured pay_001: duplicate, skipped

This time the second copy of each event says duplicate, skipped. Run the receipt count again: pay_001 now has 1. The projection shows pay_001 as captured and pay_002 as authorized, matching the payments database.

The relay isn't your only source of repeats. Kafka also redelivers whatever a consumer handled but didn't commit. Restart the notifier with a crash staged between the database commit and the offset commit, then make a new payment:

docker stop notifier && docker rm notifier
docker compose run -d --name notifier -e CRASH_AT=before_offset_commit app python notifier.py
docker compose run --rm app python payments.py authorize pay_004 1500
docker compose run --rm app python payments.py capture pay_004

Wait about 10 seconds, then read the log:

docker logs notifier

The log ends with crash at before_offset_commit: the notifier recorded payment.authorized pay_004 in its database and died before telling Kafka. Start a fresh notifier:

docker rm notifier
docker compose run -d --name notifier app python notifier.py

The new consumer waits until Kafka gives up on the dead one. The classic group protocol, still librdkafka's default, has a 45-second session timeout; it took about 20 seconds in my run. Then read the log:

docker logs notifier

Output

p0@2 payment.authorized pay_004: duplicate, skipped
p0@3 payment.captured pay_004: receipt sent

Kafka redelivered offset 2, the inbox recognized it, and the capture went through once. Same inbox, different failure. Leave this notifier running.

Step 7: Keep Per-Payment Order with One Active Relay

One relay is a single point of failure, so the natural next move is to run two, and SKIP LOCKED lets them share the table. To see what that does to order, replace the relay with one that holds each batch for 6 seconds before producing:

docker stop relay && docker rm relay
docker compose run -d --name relayA -e SINGLE_ACTIVE=0 -e HOLD_SECONDS=6 app python relay.py

Now authorize and capture a payment, and start a second relay, inside those 6 seconds. Chaining the commands keeps the timing reliable:

docker compose run --rm app python payments.py authorize pay_005 3000 && docker compose run --rm app python payments.py capture pay_005 && docker compose run -d --name relayB -e SINGLE_ACTIVE=0 app python relay.py

relayA locks payment.authorized pay_005 and sits on it. relayB skips that locked row, finds payment.captured pay_005, and publishes it at once. Wait about 10 seconds, then check the notifier:

docker logs notifier

The last two lines will look similar to these:

Output

p2@0 payment.captured pay_005: receipt sent
p2@1 payment.authorized pay_005: status updated

The capture arrived before the authorization. Compare the two databases:

docker compose exec postgres psql -U app -d payments -c "SELECT id, status FROM payments WHERE id = 'pay_005';"
docker compose exec postgres psql -U app -d notifier -c "SELECT * FROM payment_status WHERE payment_id = 'pay_005';"

The payments database says captured; the notifier says authorized. Kafka kept the order the relays gave it, and they gave it the wrong one. The inbox can't help: these are two different events.

The fix is to let one relay publish and keep the others on standby. relay.py does that by default with pg_try_advisory_lock, which returns false at once if another session holds the lock on RELAY_LOCK. It's a session-level lock, held until released or the session ends and not tied to any transaction, so it outlives each batch and vanishes with the holder's connection.

Session locks need a session

A connection pooler in transaction mode, such as PgBouncer with transaction pooling, hands each transaction to whichever server connection is free. From the documentation's description of session locks, I'd expect such a pooler to break this lock: it can be taken on one server connection and the next batch can run on another. Give the relay a direct or session-pooled connection.

Restart both relays in their default single-active mode, keeping relayA slow:

docker rm -f relayA relayB
docker compose run -d --name relayA -e HOLD_SECONDS=6 app python relay.py
docker compose run -d --name relayB app python relay.py

Wait about 10 seconds, then read relayB's log with docker logs relayB. It prints standby: another relay is active every 2 seconds. Authorize and capture another payment:

docker compose run --rm app python payments.py authorize pay_006 3000
docker compose run --rm app python payments.py capture pay_006

After relayA's two 6-second holds, docker logs notifier shows payment.authorized pay_006 and then payment.captured pay_006. Now kill the active relay and give the standby work:

docker kill relayA
docker compose run --rm app python payments.py authorize pay_007 2000
docker logs relayB

relayB prints active within one 2-second retry and publishes pay_007: PostgreSQL released the lock when the killed process's connection closed.

That's the forgiving failure. A relay cut off by a network partition is noticed only after TCP timeouts or keepalives, and may keep producing meanwhile. I didn't test that case; it's the lease problem from Distributed Locks in Redis with Go, where a lease can't prove the old owner stopped.

I'd run one active relay and shard relays by aggregate_id only when the lag query in Step 9 says one can't keep up.

Remove both relays before the next step:

docker rm -f relayA relayB

Step 8: Run a Chaos Test

So far each failure happened once, on cue. This step runs them all at random, under load. loadgen.py authorizes and captures payments from 8 threads:

loadgen.py

import sys
from concurrent.futures import ThreadPoolExecutor

import psycopg

from common import PAYMENTS_DB
from payments import authorize, capture


def one(n: int) -> None:
    payment_id = f"pay_{n:05d}"
    with psycopg.connect(PAYMENTS_DB, autocommit=True) as conn:
        authorize(conn, payment_id, 100 + n)
        capture(conn, payment_id)


count = int(sys.argv[1])
with ThreadPoolExecutor(max_workers=8) as pool:
    list(pool.map(one, range(count)))
print(f"{count} payments authorized and captured")

verify.py compares the payments database, the topic's end offsets and the notifier's database:

verify.py

import psycopg
from confluent_kafka import Consumer, TopicPartition

from common import KAFKA, NOTIFIER_DB, PAYMENTS_DB, TOPIC

with psycopg.connect(PAYMENTS_DB) as conn:
    captured, events, unpublished = conn.execute(
        """SELECT (SELECT count(*) FROM payments WHERE status = 'captured'),
                  (SELECT count(*) FROM outbox),
                  (SELECT count(*) FROM outbox WHERE published_at IS NULL)"""
    ).fetchone()

probe = Consumer({"bootstrap.servers": KAFKA, "group.id": "verify"})
on_topic = sum(probe.get_watermark_offsets(TopicPartition(TOPIC, p), timeout=10)[1] for p in range(3))
probe.close()

with psycopg.connect(NOTIFIER_DB) as conn:
    processed, receipts, paid_once, status_captured = conn.execute(
        """SELECT (SELECT count(*) FROM processed_events),
                  (SELECT count(*) FROM receipts_sent),
                  (SELECT count(DISTINCT payment_id) FROM receipts_sent),
                  (SELECT count(*) FROM payment_status WHERE status = 'captured')"""
    ).fetchone()

print(f"payments captured:          {captured}")
print(f"outbox events:              {events} ({unpublished} unpublished)")
print(f"messages on the topic:      {on_topic} ({on_topic - events} duplicates)")
print(f"events processed once:      {processed}")
print(f"receipts sent:              {receipts} for {paid_once} payments")
print(f"projection shows captured:  {status_captured}")

Reset everything so every count belongs to this run:

docker stop notifier && docker rm notifier
docker compose exec postgres psql -U app -d payments -c "TRUNCATE payments, outbox RESTART IDENTITY;"
docker compose exec postgres psql -U app -d notifier -c "TRUNCATE processed_events, receipts_sent, payment_status;"
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
docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group notifier

Start two relays and a notifier, each in a restart loop. The relays crash after producing on 10% of batches, the notifier before its offset commit on 0.2% of events:

docker compose run -d --name relayA -e CRASH_AT=after_produce -e CRASH_RATE=0.1 app sh -c 'while true; do python relay.py; done'
docker compose run -d --name relayB -e CRASH_AT=after_produce -e CRASH_RATE=0.1 app sh -c 'while true; do python relay.py; done'
docker compose run -d --name notifier -e CRASH_AT=before_offset_commit -e CRASH_RATE=0.002 app sh -c 'while true; do python notifier.py; done'

Send 1,000 payments:

docker compose run --rm app python loadgen.py 1000

In my run, the load took about 2.5 seconds and the pipeline drained within about 40, most of it one crashed notifier's session timing out. Run verify.py until it reports 0 unpublished and 2,000 processed:

docker compose run --rm app python verify.py

Output

payments captured:          1000
outbox events:              2000 (0 unpublished)
messages on the topic:      2201 (201 duplicates)
events processed once:      2000
receipts sent:              1000 for 1000 payments
projection shows captured:  1000

Notice the numbers that matter: 201 duplicates on the topic, yet 2,000 events processed once and 1,000 receipts. In the logs, relayA crashed 3 times and published 37 batches; relayB published nothing, because relayA's restart loop won the lock back each time. The notifier crashed once and skipped 202 duplicates: 201 relay repeats plus 1 redelivery. The crashes are random, so counts vary from run to run; the zeros and the thousands shouldn't.

With a single active relay, you might expect the topic to be in order. replay_order.py reads the whole topic and counts payments whose authorized event appears after their captured one:

replay_order.py

import json

from confluent_kafka import Consumer

from common import KAFKA, TOPIC

consumer = Consumer({"bootstrap.servers": KAFKA, "group.id": "audit", "auto.offset.reset": "earliest", "enable.auto.commit": False})
consumer.subscribe([TOPIC])
captured, stale, idle = set(), set(), 0
while idle < 5:
    msg = consumer.poll(1.0)
    if msg is None:
        idle += 1
        continue
    idle = 0
    event = json.loads(msg.value())
    if event["type"] == "payment.captured":
        captured.add(event["payment_id"])
    elif event["payment_id"] in captured:
        stale.add(event["payment_id"])
consumer.close()
print(f"payments with an 'authorized' arriving after 'captured': {len(stale)}")

Then run:

docker compose run --rm app python replay_order.py

Output

payments with an 'authorized' arriving after 'captured': 98

This was the result that surprised me in the chaos run: 98 payments, with only one relay active at a time. When a relay crashes after producing, its rollback and the release of its lock happen together, and the next relay to take the lock selects the same oldest unpublished rows. The crashed batch is published again whole, straight after its first copy, so a payment whose authorize and capture were both in that batch reads authorized, captured, authorized, captured on the topic. Order on the topic is the order of first delivery only.

The inbox skipped every replayed event by its event_id. Without it, those payments would flip back to authorized until the replayed capture arrived, and each customer would get a second receipt.

Step 9: Monitor and Clean Up the Outbox

You still need to know when the relay falls behind, and the outbox grows forever unless you clean it. ops.sql handles both:

ops.sql

-- How far behind is the relay?
SELECT count(*) AS unpublished,
       coalesce(max(now() - created_at), interval '0') AS oldest_unpublished_age
FROM outbox
WHERE published_at IS NULL;

-- Delete published events older than 7 days, 10,000 rows per run.
DELETE FROM outbox
WHERE id IN (
    SELECT id FROM outbox
    WHERE published_at < now() - interval '7 days'
    ORDER BY id
    LIMIT 10000
);

Every outbox row is minutes old, so the DELETE would find nothing. To see it work, backdate the first 500 published events (test only):

docker compose exec postgres psql -U app -d payments -c "UPDATE outbox SET published_at = now() - interval '8 days' WHERE id <= 500;"

Output

UPDATE 500

Run ops.sql against the payments database. -T stops Compose from allocating a terminal, because input comes from a file:

docker compose exec -T postgres psql -U app -d payments < ops.sql

Output

 unpublished | oldest_unpublished_age 
-------------+------------------------
           0 | 00:00:00
(1 row)

DELETE 500

The relay is fully caught up, and the 500 backdated rows are gone. I'd alert on oldest_unpublished_age rather than the count: a thousand events a second old is a busy minute, while one event ten minutes old means the relay is down or stuck. Deleting in capped batches keeps each run short; schedule it, and size the 7 days to how long you might need to replay or inspect events.

To confirm the relay's query uses the partial index, run it with EXPLAIN:

docker compose exec postgres psql -U app -d payments -c "EXPLAIN SELECT id, event_id, aggregate_id, event_type, payload, created_at FROM outbox WHERE published_at IS NULL ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED;"

Output

                                             QUERY PLAN                                             
----------------------------------------------------------------------------------------------------
 Limit  (cost=77.92..78.31 rows=31 width=134)
   ->  LockRows  (cost=77.92..78.31 rows=31 width=134)
         ->  Sort  (cost=77.92..78.00 rows=31 width=134)
               Sort Key: id
               ->  Bitmap Heap Scan on outbox  (cost=4.44..77.15 rows=31 width=134)
                     Recheck Cond: (published_at IS NULL)
                     ->  Bitmap Index Scan on outbox_unpublished  (cost=0.00..4.43 rows=31 width=0)
(7 rows)

The plan reads outbox_unpublished. Depending on table statistics, PostgreSQL uses it through a Bitmap Index Scan, as here, or a plain Index Scan; your costs will differ. A sequential scan on outbox would mean the index isn't being used.

When you're done, remove the lab:

docker rm -f relayA relayB notifier
docker compose down -v

Polling Relay vs Debezium CDC

Polling isn't the only way to get rows out of the outbox. The main alternative is change data capture (CDC): Debezium's PostgreSQL connector reads the write-ahead log, PostgreSQL's record of every change, through the pgoutput plugin, and its Outbox Event Router turns outbox inserts into Kafka messages. The latest release at the time of writing is Debezium 3.7.0.Final.

	Polling relay (this tutorial)	Debezium with the Outbox Event Router
How it finds events	`SELECT ... FOR UPDATE SKIP LOCKED` every 0.5 s when idle	Reads the write-ahead log via `pgoutput`
What you run	One small Python process	Kafka Connect plus the Debezium connector
Load on the database	Polling queries and an `UPDATE` per batch	No polling queries
Outbox columns	Whatever you choose	Defaults: `id`, `aggregatetype`, `aggregateid`, `type`, `payload`
Message key	You set it (`aggregate_id` here)	`aggregateid` by default
Delivery	At least once	At least once by default; optional exactly-once into Kafka through Kafka Connect (`exactly.once.source.support=enabled` on the workers, `exactly.once.support=required` on the connector)

Neither removes the consumer's inbox. Debezium's exactly-once mode covers writes into Kafka, and its own documentation says it remains unclear whether that implementation is fully correct; a consumer that crashes before committing its offset is sent the message again either way. If you already run Kafka Connect, Debezium spares you the polling and the relay code. If you don't, I'd start with the polling relay: it's about 70 lines you can read in one sitting, and moving to CDC later only changes how events leave the table.

Troubleshooting

Symptom	Cause	Fix
`Failed to acquire idempotence PID from broker ... Coordinator load in progress: retrying` (a `GETPID` warning)	The idempotent producer asked a freshly started broker for a producer ID before its coordinator was ready	Nothing; librdkafka retries on its own
`Assignments can only be reset if the group 'notifier' is inactive, but the current state is Stable`	A consumer is still a member of the group, often a killed one whose session hasn't timed out	Stop consumers with `docker stop`, or wait out the session timeout, then reset
The group keeps a member after `docker stop`, and the next consumer waits	Python runs as the container's first process, which ignores `SIGTERM` unless it installs a handler, so `docker stop` sends `SIGKILL` after 10 seconds; `finally` never runs and `consumer.close()` never tells Kafka the consumer left	Install a `SIGTERM` handler that calls `sys.exit(0)`, as `notifier.py` does. With it, the group shows `Empty` with 0 members after a stop
`Option --property is deprecated ... Use --formatter-property instead`	Kafka 4.3 prints a deprecation warning for `--property` on the console consumer	Use `--formatter-property print.key=true`
A Java service and a Python service put the same key in different partitions	librdkafka's default partitioner is `consistent_random` (CRC32); the Java producer's default hashes keys with murmur2	Set `"partitioner": "murmur2_random"` in every librdkafka producer that shares a topic with Java producers; librdkafka documents it as functionally equivalent to Java's default
Events are never published, with no error	The relay tracks `id > last_seen`, and a slow transaction committed a lower id after the cursor passed it	Poll `published_at IS NULL`, as in Step 4

Conclusion

You built the transactional outbox pattern in Python: the outbox ties each event to its payment's transaction, and publishing is at-least-once whatever the producer's settings. The session advisory lock stops two relays from reordering events, and the inbox absorbs the reordering that replays still cause; together they turned 1,000 payments and 201 duplicates into exactly 1,000 receipts.

For the people deciding whether it's worth it: the outbox costs a table, a small relay process with a standby, an inbox table and an offset rule in every consumer, a cleanup job and one lag alert. In return, no event depends on a process surviving between two writes. Ask any team that publishes from inside a request handler what happens when the process dies between the commit and the publish; often the answer is reconciliation by hand.

Next, read the pattern on microservices.io and AWS Prescriptive Guidance, and cover the request side with idempotency keys in Go and PostgreSQL. Then run Step 8's chaos test against your own consumers.

If your team is moving events between services and wants the delivery guarantees checked, I can review the design and build the tests that check it. You can see how I approach distributed systems, or book a free 20-minute intro call.

Originally published at shahidyousuf.dev by Shahid Yousuf.