librdkafka

confluent-kafka-go in 2026: What Still Works and What to Watch confluent-kafka-go is Confluent's Kafka client for Go. It wraps librdkafka, Confluent's C client, through cgo, and it is still actively developed. v2.16.0 shipped on 7 October 2026. If you run a Go service on it, five things deserve a check: how it builds, what its Schema Registry packages pull in, where keyed records land, when offsets get committed, and which consumer group protocol it uses. TL;DR: confluent-kafka-go covers transactions, admin operations, KIP-848 consumer groups, and Confluent's own Schema Registry serializers. It needs cgo, and Go turns cgo off when you cross-compile. The Schema Registry serializers need cgo too, even next to a pure Go client. The default partitioner does not match the Java client, but murmur2_random does. By default, offsets can be committed before your code finishes processing a message. The consumer group protocol is expected to default to KIP-848 in a later release, so set it explicitly. For how confluent-kafka-go compares with franz-go, kafka-go, and Sarama on cgo, consumer groups, and Schema Registry, see Kafka Go Clients: franz-go vs kafka-go vs Sarama. What still works - librdkafka underneath. Most protocol behavior comes from the bundled librdkafka, the same C library behind Confluent's Python, .NET, and JavaScript clients. v2.16.0 bundles librdkafka 2.16.0, with idempotent and transactional producers and an AdminClient for topics, configs, ACLs, and consumer groups. - No separate install on common platforms. The module ships prebuilt librdkafka for macOS (x64 and arm64), glibc Linux (x64 and arm64), musl Linux such as Alpine (x64 and arm64), and Windows x64. On those platforms, you do not install librdkafka on the build or target machine. - KIP-848 consumer groups. The next-generation consumer group protocol, in which the broker computes partition assignments, has been production-ready since v2.12.0 on Kafka 4.0+ brokers. You turn it on with group.protocol=consumer. - Confluent's Schema Registry serializers. The schemaregistry packages include serializers for Avro, Protobuf, and JSON Schema that register and cache schemas and write Confluent's wire format. - Active development. In the 90 days to 25 September 2026, the repository had 55 commits and 29 merged pull requests. Over the preceding year, merges were spread across 17 accounts, and the most active one handled 39% of them. The open backlog is older, with 74 open pull requests at a median age of 2.4 years. These figures come from the GitHub snapshot in the full comparison. What to watch cgo is required, and cross-compiling turns it off The client calls librdkafka through cgo, so it builds only with CGO_ENABLED=1 and a C compiler. Go turns cgo off by default when you cross-compile, for example GOOS=linux go build on a Mac, and when it cannot find a C compiler, as in the golang:alpine image. With cgo off, a v2.16.0 build fails inside the library: # github.com/confluentinc/confluent-kafka-go/v2/kafka .../kafka/deserialized_message.go:28:17: undefined: TopicPartition .../kafka/deserialized_message.go:34:17: undefined: TimestampType .../kafka/deserialized_message.go:36:19: undefined: Header Up to v2.15.1, the same build fails in your own code instead, with errors such as undefined: kafka.NewProducer and undefined: kafka.ConfigMap. To build for another platform, set CGO_ENABLED=1 and give cgo a C compiler for that platform, or build on the target platform itself, for example in a container for that architecture. A few more build details matter once the service ships in a container: - Alpine needs -tags musl. The project README requires it for go get, go build, and similar commands on musl systems. - The binary is not static. It links the system C library dynamically, so the runtime image needs one. scratch and distroless/static have none, and a binary built against a newer glibc fails on an older one. For a fully static binary, the project README describes a musl build with external-linker flags. - Kerberos needs a system librdkafka. The bundled builds exclude GSSAPI. Install a librdkafka with GSSAPI on both the build and target systems, and build with -tags dynamic. Without librdkafka and pkg-config on the build machine, that build stops with exec: "pkg-config": executable file not found in $PATH. - v2.16.0 needs Go 1.26. The module's go directive moved from 1.25.0 to 1.26.0. With the default GOTOOLCHAIN=auto, Go 1.25 downloads a newer toolchain by itself. With GOTOOLCHAIN=local, you need to upgrade Go first. Schema Registry serializers need cgo too The Schema Registry client and serializers do not call librdkafka, so it seems natural to pair them with a pure Go client such as franz-go and build with CGO_ENABLED=0. For the serializers, that stopped working in v2.10.1 (June 2025). In that release, the serializer interface gained methods that can carry the schema ID in Kafka headers. Those methods use the kafka.Header type, and the kafka package is the one that needs cgo. We built a small program that uses the Avro serializer (avrov2) with franz-go, and a second one that uses only the Registry client, with CGO_ENABLED=0 and Go 1.27.1 against each release: | Release | Registry client only | Avro serializer | | ------------------ | --------------------------------- | ------------------------- | | v2.5.4 to v2.10.0 | Builds | Builds | | v2.10.1 to v2.13.0 | Builds | undefined: kafka.Header | | v2.13.2 to v2.15.1 | undefined: kafka.LibraryVersion | Same error | | v2.16.0 | undefined: TopicPartition | Same error | The v2.16.0 release notes say this dependency was removed from the Registry client, and that change did ship. The same release, however, added an experimental SerializingProducer and DeserializingConsumer API. It brought a new import of the kafka package into the Registry client, and new files in the kafka package that do not compile without cgo. Both problems are tracked in issue #1635, with a proposed fix in pull request #1634. The release notes also mark that new API as subject to change or removal, so pin the module version if you adopt it. The default partitioner does not match Java The default partitioner, consistent_random, hashes keys with CRC32, while the Java client and franz-go use Murmur2. If a Go service and a Java service write the same keys to one topic, a key can land on two partitions, which breaks per-key ordering. Setting partitioner to murmur2_random gives Java-compatible placement: p, err := kafka.NewProducer(&kafka.ConfigMap{ "bootstrap.servers": "localhost:9092", "partitioner": "murmur2_random", }) We checked it by producing with v2.16.0 to a local Kafka 4.3.1 broker and comparing the partition in each delivery report with the Java client 4.3.1 (BuiltInPartitioner.partitionForKey). We used the keys customer_42, order-1001, user-7, 21, foobar, a-little-bit-long-string, café, and an empty key, each at 3, 6, and 12 partitions. murmur2_random matched Java in all 24 cases. The default matched in only 1 of the 21 cases with a non-empty key. For example, on a three-partition topic customer_42 goes to partition 1 by default and to partition 0 with murmur2_random, as in Java. Empty keys behave differently too. The default sends an empty key to a random partition. Five sends to a 12-partition topic landed on partitions 4, 0, 9, 7, and 0. murmur2_random hashes it as Java does and sent it to partition 9 every time. Records with a null key go to a random partition under both settings. After a switch, new records for a key can go to a different partition than its earlier records. Switch when other producers write the same keys to the same topic, and compare placements with your own keys first. Offsets can be committed before processing finishes By default, the consumer stores a message's offset as soon as Poll or ReadMessage hands the message to your code (enable.auto.offset.store=true). It then commits the stored offsets in the background every 5 seconds (enable.auto.commit=true, auto.commit.interval.ms=5000). If the process crashes after a commit but before your code finishes a message, that message is not delivered again. For at-least-once processing, turn off the automatic store and store each offset yourself after the work is done. Auto-commit stays on and commits only what you stored: c, err := kafka.NewConsumer(&kafka.ConfigMap{ "bootstrap.servers": "localhost:9092", "group.id": "orders", "enable.auto.offset.store": false, }) // After processing msg: if _, err := c.StoreMessage(msg); err != nil { // handle the error } This works when you process each partition's messages in order. If you process one partition's messages in parallel, store an offset only after every earlier message in that partition has succeeded. After a crash, messages that finished but were not yet committed are delivered again, so processing must tolerate duplicates. KIP-848 is opt-in for now The default group.protocol is still classic, but librdkafka's configuration reference says the default "will change to consumer in next releases." The new protocol changes a few contracts. The client no longer uses partition.assignment.strategy, session.timeout.ms, or heartbeat.interval.ms. Rebalance callbacks receive only the partitions being added or revoked. If your callbacks manage assignments themselves, use IncrementalAssign and IncrementalUnassign instead of Assign and Unassign. Set group.protocol explicitly, so that a future upgrade does not switch protocols for you. Before moving a group to consumer, read Confluent's KIP-848 migration guide. If you need what it lacks A service that builds and runs well on confluent-kafka-go can keep running it. If cgo is the problem, franz-go is a pure Go client that supports KIP-848. Its own Schema Registry package, pkg/sr, writes the same wire format, but you supply the Avro, Protobuf, or JSON Schema codec. The full comparison linked above covers what to test before switching, including how to keep key placement. Check what your service actually does Whether you change the partitioner, the offset store, or the build image, check the records themselves. Kafma, a Kafka GUI client, shows each record's key, partition, headers, and raw bytes, and a consumer group's committed offsets and lag. That lets you confirm that keys land where you expect. To check whether commits run ahead of processing, compare the committed offsets with your application logs. Rebalances, retries, and crash recovery still need tests in your application. Written by the team behind Kafma.

How to Implement the Transactional Outbox Pattern in Python --- title: How to Implement the Transactional Outbox Pattern in Python published: true date: 2026-10-19 14:30:00 UTC tags: DistributedSystems,Kafka,PostgreSQL,Python canonical_url: shahidyousuf.dev cover_image: shahidyousuf.dev published_at: "2026-10-21T04:34:51Z" --- 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 rel