In modern microservices and event-driven architectures, transactional data rarely lives in a single database. An e-commerce order must be stored in an ACID relational database (PostgreSQL/MySQL), indexed in a search engine (Elasticsearch/OpenSearch), cached in memory (Redis), and broadcast to downstream microservices (Inventory, Billing, Analytics) via an event broker (Apache Kafka).
How do you synchronize these disparate storage engines reliably in real time? The naive approach—having the application write to the database and immediately publish a message to Kafka (the Dual-Write Anti-Pattern)—inevitably leads to catastrophic data corruption and state drift during network partitions or process crashes.
The industry standard solution is Change Data Capture (CDC). By mining the database's internal transaction log (PostgreSQL's Write-Ahead Log or MySQL's Binlog) at the storage engine level, CDC extracts committed row-level mutations (INSERT, UPDATE, DELETE) as a continuous event stream with zero application overhead, 100% data consistency, and sub-second latency.
In this technical deep dive, we explore the dual-write problem, dissect PostgreSQL Logical Decoding (pgoutput) and Replication Slots, examine Debezium event formats, master the Transactional Outbox Pattern, and build an end-to-end CDC synchronization pipeline in Go.
The Dual-Write Problem: Why Application-Level Sync Fails
Consider what happens when an application service attempts to write to both a database and a message broker within the same business transaction:
The Dual-Write Race Condition:
Scenario 1: DB Succeeds, Kafka Fails
1. App executes: BEGIN; INSERT INTO orders; COMMIT; -> (DB Success!)
2. App calls: kafkaProducer.send(orderCreatedEvent);
3. Network glitch / Kafka timeout occurs -> (Message FAILS!)
RESULT: Database has order, but Inventory service never receives the event! (Ghost State)
Scenario 2: Kafka Succeeds, DB Rolls Back
1. App calls: kafkaProducer.send(orderCreatedEvent); -> (Message Sent!)
2. App executes: INSERT INTO orders; -> (Unique Constraint Violation / Crash!)
3. DB Transaction Rolls Back!
RESULT: Downstream services process order that does not physically exist!
Scenario 3: Concurrent Out-of-Order Overwrites
Thread A: UPDATE user SET email = '[email protected]' (DB) ───[ Delayed on Wire ]───> Kafka
Thread B: UPDATE user SET email = '[email protected]' (DB) ─────────────────────────> Kafka (Arrives First!)
RESULT: Database has '[email protected]', but Kafka ends on '[email protected]'! (Permanent Drift)Because distributed two-phase commit (XA / 2PC) across relational databases and Kafka brokers is notoriously fragile and slow, application-level dual-writes are fundamentally broken.
PostgreSQL Logical Replication: Write-Ahead Log (WAL) Decoding
PostgreSQL achieves durability by appending every atomic mutation to the Write-Ahead Log (WAL) before modifying database pages on disk. The WAL is a strict, immutable chronological record of truth.
PostgreSQL Logical Replication Architecture:
┌────────────────────────────────────────────────────────┐
│ PostgreSQL Primary Instance │
│ │
│ 1. SQL Mutation (INSERT / UPDATE / DELETE) │
│ │ │
│ ▼ │
│ [ Write-Ahead Log (WAL) on Disk ] │
│ Record 1: LSN 0/16B3D40 (Table: users, Row: 101) │
│ Record 2: LSN 0/16B3E08 (Table: orders, Row: 42) │
│ │ │
│ ▼ │
│ [ Logical Decoding Plugin (pgoutput / test_decoding) ]│
│ Decodes raw WAL binary into structured row events │
│ │ │
│ ▼ │
│ [ Replication Slot: 'debezium_slot' ] │
│ Tracks Consumer LSN Position -> Retains WAL files │
└───────────┼────────────────────────────────────────────┘
│ Streaming Protocol (TCP)
▼
┌────────────────────────────────────────────────────────┐
│ Debezium / CDC Consumer │
│ Consumes JSON/Avro events -> Publishes to Kafka │
│ Acknowledges LSN -> Postgres frees old WAL segments │
└────────────────────────────────────────────────────────┘Key PostgreSQL Concepts:
wal_level = logical: Instructs PostgreSQL to include row metadata, old/new column values, and transaction boundaries in the WAL.- Logical Decoding Plugin (
pgoutput): Converts internal binary WAL records into an understandable logical replication stream (standard in PostgreSQL 10+). - Replication Slot: A persistent state tracker on the primary server that guarantees PostgreSQL will not delete or recycle WAL segments until the CDC consumer acknowledges receipt of that Log Sequence Number (LSN).
max_slot_wal_keep_size!
Debezium: The Open-Source CDC Standard
Debezium is a distributed, Kafka Connect-native platform that provides production-grade connectors for PostgreSQL, MySQL, MongoDB, SQL Server, and Oracle.
When a row is modified, Debezium emits a rich, self-describing event containing both the before and after state, transaction metadata, and source coordinates:
{
"schema": { "...": "JSONSchema / Avro schema metadata" },
"payload": {
"before": {
"id": 42,
"email": "[email protected]",
"status": "pending"
},
"after": {
"id": 42,
"email": "[email protected]",
"status": "active"
},
"source": {
"version": "2.7.0.Final",
"connector": "postgres",
"db": "production_db",
"schema": "public",
"table": "users",
"txId": 10842,
"lsn": 23828944,
"ts_ms": 1789452000000
},
"op": "u",
"ts_ms": 1789452000120
}
}Operation Code (op) |
SQL Action | Payload Content |
|---|---|---|
| c (Create) | INSERT |
before: null, after: { ... new values ... } |
| u (Update) | UPDATE |
before: { ... old ... }, after: { ... new ... } |
| d (Delete) | DELETE |
before: { ... old ... }, after: null |
| r (Read) | Initial Snapshot | Initial load records generated during initial connector sync. |
The Transactional Outbox Pattern: CDC-Powered Domain Events
While raw table CDC is powerful for data replication, publishing internal database schemas directly to external microservices tightly couples your internal tables to public API contracts.
The Transactional Outbox Pattern combines local database transactions with CDC to emit clean, business-level domain events:
The Transactional Outbox Architecture:
1. Application executes ACID Transaction in PostgreSQL:
BEGIN;
INSERT INTO orders (id, user_id, total) VALUES ('ord_1', 'usr_42', 99.00);
-- Write domain event into the dedicated Outbox table in the SAME transaction!
INSERT INTO outbox_events (
id, aggregate_type, aggregate_id, event_type, payload
) VALUES (
'evt_99', 'Order', 'ord_1', 'OrderCreated', '{"userId": "usr_42", "total": 99.00}'
);
COMMIT;
(Atomic guarantee: Either BOTH write or NEITHER writes!)
2. Debezium captures 'outbox_events' table via Postgres WAL.
3. Debezium Outbox Event Router re-routes event to Kafka topic: 'events.orders.OrderCreated'.
4. Debezium auto-deletes the processed outbox row (or marks it published).Hands-On Implementation: Postgres Logical Replication in Go
You don't always need a heavy Java/Kafka-Connect cluster. You can consume PostgreSQL logical replication streams directly in Go using jackc/pglogrepl:
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/jackc/pglogrepl"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgproto3"
)
const (
DB_CONN = "postgres://postgres:secret@localhost:5432/app_db?replication=database"
SLOT = "go_cdc_slot"
PLUGIN = "pgoutput"
)
func main() {
ctx := context.Background()
// 1. Establish replication connection to PostgreSQL
conn, err := pgconn.Connect(ctx, DB_CONN)
if err != nil {
log.Fatalf("Failed to connect to PostgreSQL: %v", err)
}
defer conn.Close(ctx)
// 2. Create Logical Replication Slot (if not exists)
_, err = pglogrepl.CreateReplicationSlot(ctx, conn, SLOT, PLUGIN, pglogrepl.CreateReplicationSlotOptions{Temporary: false})
if err != nil {
log.Printf("Slot may already exist: %v", err)
}
// 3. Start Replication Stream
sysident, err := pglogrepl.IdentifySystem(ctx, conn)
if err != nil {
log.Fatalf("IdentifySystem failed: %v", err)
}
log.Printf("Postgres System ID: %s, Current XLogPos: %s", sysident.SystemID, sysident.XLogPos)
err = pglogrepl.StartReplication(ctx, conn, SLOT, sysident.XLogPos, pglogrepl.StartReplicationOptions{
PluginArgs: []string{"proto_version '1'", "publication_names 'cdc_publication'"},
})
if err != nil {
log.Fatalf("StartReplication failed: %v", err)
}
log.Println("Listening for live PostgreSQL WAL changes...")
// 4. Ingestion Loop
standbyMessageTimeout := 10 * time.Second
nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout)
for {
if time.Now().After(nextStandbyMessageDeadline) {
// Send Standby Status Update heartbeat to Postgres to acknowledge processed LSN
err = pglogrepl.SendStandbyStatusUpdate(ctx, conn, pglogrepl.StandbyStatusUpdate{
WALWritePosition: sysident.XLogPos,
})
if err != nil {
log.Fatalf("SendStandbyStatusUpdate failed: %v", err)
}
nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
}
ctxTimeout, cancel := context.WithDeadline(ctx, nextStandbyMessageDeadline)
rawMsg, err := conn.ReceiveMessage(ctxTimeout)
cancel()
if err != nil {
if pgconn.Timeout(err) {
continue
}
log.Fatalf("ReceiveMessage failed: %v", err)
}
if msg, ok := rawMsg.(*pgproto3.CopyData); ok {
switch msg.Data[0] {
case pglogrepl.PrimaryKeepaliveMessageByte:
pkm, _ := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
if pkm.ReplyRequested {
nextStandbyMessageDeadline = time.Time{}
}
case pglogrepl.XLogDataByte:
xld, _ := pglogrepl.ParseXLogData(msg.Data[1:])
fmt.Printf("[CDC Event] LSN: %s | Bytes: %d\n", xld.WALStart, len(xld.WALData))
// Forward decoded WAL event to Elasticsearch / Redis / Kafka!
}
}
}
}Summary & Engineering Checklist
- Eliminate Dual-Writes: Never write to databases and message queues sequentially in user-space code. Standardize on CDC or the Transactional Outbox pattern.
- Monitor Replication Slots Vigilantly: Implement Prometheus metrics on
pg_replication_slots.wal_statusand replication lag bytes to prevent disk exhaustion. - Decouple Schemas with the Outbox Pattern: Use dedicated outbox tables to publish curated domain event schemas rather than leaking raw database table structures to downstream consumers.
- Ensure Idempotent Consumers: Because network reconnections can deliver duplicate change events during failovers (At-Least-Once delivery), all downstream consumers (Elasticsearch/Redis) must process events idempotently using natural database keys or Log Sequence Numbers (LSN).