Building a Simple Change Data Capture Pipeline With Debezium and Go

Real-time replication from OLTP to analytics — minimal setup

A few weeks ago I had one of those “why didn’t we do this earlier” moments. Our analytics team kept asking for fresher data from our main Postgres database. The existing solution was a cron job that ran every hour, dumped a few tables, and pushed them into our warehouse. It worked, but it was clunky, it lagged, and every time someone added a column to the source table, the export job quietly broke.

So I spent a weekend setting up Change Data Capture (CDC) with Debezium, Kafka, and a small Go consumer. It turned out to be way less intimidating than I expected, and I want to walk through how I set it up — including the parts that tripped me up.

Why CDC instead of batch exports

Batch exports have a few annoying properties:

• They’re either too frequent (wasting resources) or too infrequent (stale data)
• They usually do full table scans, which gets expensive as tables grow
• Schema changes break them in subtle ways
• They don’t capture deletes well unless you build extra logic for it

CDC flips this around. Instead of asking “what changed since last time?”, you tap directly into the database’s write-ahead log (WAL) and get a stream of every insert, update, and delete as it happens. Debezium does the heavy lifting of reading that WAL and turning it into structured events on Kafka topics.

The architecture

Here’s the rough shape of what I built:

Postgres (OLTP) --> Debezium Connector --> Kafka --> Go Consumer --> Analytics DB (e.g. ClickHouse/BigQuery)

A simplified view of the data flow:

+-------------+        +-----------+        +-------+        +-------------+        +------------+
|  Postgres   |  WAL   |  Debezium |  CDC   | Kafka |  Sub   |  Go Consumer |  Write |  Analytics |
|  (orders)   +------->+ Connector +------->+ Topic +------->+  Service     +------->+    DB      |
+-------------+        +-----------+        +-------+        +-------------+        +------------+

Throughput-wise, on a modest setup (a small Postgres instance with moderate write volume), I was comfortably seeing replication lag in the low hundreds of milliseconds — a massive improvement over the hourly batch job.

Step 1: Enable logical replication on Postgres

Debezium relies on Postgres’s logical decoding feature, so the first step is making sure your postgresql.conf has:

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

Then create a replication user and a publication for the tables you care about:

CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'debezium';

CREATE PUBLICATION dbz_publication FOR TABLE orders, order_items;

Restart Postgres for the WAL settings to take effect.

Step 2: Spin up Debezium with Docker Compose

For local testing, this docker-compose.yml got me up and running quickly:

version: "3.7"
services:
  zookeeper:
    image: quay.io/debezium/zookeeper:2.5
    ports:
      - "2181:2181"

  kafka:
    image: quay.io/debezium/kafka:2.5
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      ZOOKEEPER_CONNECT: zookeeper:2181

  connect:
    image: quay.io/debezium/connect:2.5
    ports:
      - "8083:8083"
    depends_on:
      - kafka
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses

Once that’s up, register the Postgres connector via the REST API:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "debezium",
      "database.dbname": "shopdb",
      "topic.prefix": "shop",
      "table.include.list": "public.orders,public.order_items",
      "plugin.name": "pgoutput",
      "publication.name": "dbz_publication"
    }
  }'

After this, every change on orders and order_items starts flowing into Kafka topics named shop.public.orders and shop.public.order_items.

Step 3: A minimal Go consumer

This is the part I enjoyed most. I used confluent-kafka-go to consume the CDC events and write them into our analytics store. Here's a trimmed-down version focused on the orders table:

package main

import (
   "context"
   "encoding/json"
   "log"

   "github.com/confluentinc/confluent-kafka-go/v2/kafka"
)

// DebeziumEnvelope represents the structure Debezium wraps every change event in.
type DebeziumEnvelope struct {
   Payload struct {
    Before json.RawMessage `json:"before"`
    After  json.RawMessage `json:"after"`
    Op     string          `json:"op"` // "c" = create, "u" = update, "d" = delete
    Source struct {
       Table string `json:"table"`
       TsMs  int64  `json:"ts_ms"`
    } `json:"source"`
   } `json:"payload"`
}

type Order struct {
   ID         int64   `json:"id"`
   CustomerID int64   `json:"customer_id"`
   Status     string  `json:"status"`
   Total      float64 `json:"total"`
}

func main() {
   consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
    "bootstrap.servers": "localhost:9092",
    "group.id":          "analytics-sync",
    "auto.offset.reset": "earliest",
   })
   if err != nil {
    log.Fatalf("failed to create consumer: %v", err)
   }
   defer consumer.Close()

   if err := consumer.Subscribe("shop.public.orders", nil); err != nil {
    log.Fatalf("failed to subscribe: %v", err)
   }

   log.Println("listening for order changes...")

   for {
    msg, err := consumer.ReadMessage(-1)
    if err != nil {
     log.Printf("read error: %v", err)
     continue
    }

    if err := handleEvent(context.Background(), msg.Value); err != nil {
     log.Printf("failed to handle event: %v", err)
    }
   }
}

func handleEvent(ctx context.Context, raw []byte) error {
   var envelope DebeziumEnvelope
   if err := json.Unmarshal(raw, &envelope); err != nil {
    return err
   }

   switch envelope.Payload.Op {
   case "c", "u":
    var order Order
    if err := json.Unmarshal(envelope.Payload.After, &order); err != nil {
     return err
    }
    return upsertOrder(ctx, order)

   case "d":
    var order Order
    if err := json.Unmarshal(envelope.Payload.Before, &order); err != nil {
     return err
    }
    return deleteOrder(ctx, order.ID)
   }

   return nil
}

func upsertOrder(ctx context.Context, o Order) error {
   // In a real implementation, this would write to ClickHouse, BigQuery,
   // or whatever your analytics store is.
   log.Printf("upsert order: id=%d status=%s total=%.2f", o.ID, o.Status, o.Total)
   return nil
}

func deleteOrder(ctx context.Context, id int64) error {
 log.Printf("delete order: id=%d", id)
 return nil
}

A few notes from getting this running:

• The before/after fields are only populated depending on the operation type — before is empty on inserts, after is empty on deletes.
• Debezium events are idempotent-friendly if you design your sink with upserts. I went with INSERT ... ON CONFLICT DO UPDATE style writes on the analytics side, which made retries painless.
• Commit offsets only after a successful write to the analytics store, otherwise you risk losing events on a crash.

Lag monitoring

One thing worth tracking from day one is replication lag — the gap between when a change happens in Postgres and when it lands in your analytics store. I used a simple Go goroutine that periodically computed time.Now() - source.ts_ms from incoming events and pushed it to Prometheus. A rough ASCII view of what that looked like over a test run with synthetic load:

Lag (ms)
600 |                          *
500 |                    *
400 |              *
300 |        *
200 |  *
100 |*
  0 +------------------------------------
     0    5   10   15   20   25   30  (seconds)

The spike around the middle was from a burst of bulk inserts during a test seed script — normal traffic stayed well under 100ms.

What I’d do differently next time

If I were starting from scratch again, I’d skip the raw Kafka Connect REST setup and look at something like Debezium Server or a managed CDC service, especially for production. The Docker Compose setup above is great for prototyping but you’ll want proper monitoring, dead-letter queues, and schema registry integration before relying on this for anything business-critical.

That said, for a weekend project to prove out real-time analytics replication, this stack got the job done with surprisingly little code. The Go consumer ended up being under 150 lines, and most of the complexity lived in the Debezium connector configuration rather than custom code — which felt like the right tradeoff.

If you’re dealing with stale analytics data and your current pipeline is “run a script every hour and hope,” CDC is worth the afternoon it takes to try.