Building a Simple CDC Ingestion Pattern: A Practical Guide

Most data pipelines I have worked on start with a simple batch pull — run a query, dump the results, load them somewhere. But at some point, when a table grows to a few hundred million rows scraping the entire thing every hour stops making sense. That is where CDC, or Change Data Capture, comes in.

This article walks through building a simple CDC ingestion pattern from a PostgreSQL database into a data lake (in our case, GCS) using Debezium and Kafka Connect. We will not go deep into Kafka internals — the goal here is to get the pipeline running with enough understanding to debug it when things go wrong.

Why CDC Instead of Batch Pulls

A full table scan every run works until it does not. The problems are not just about cost, though scanning a 500 GB table on every pipeline run will show up on your bill. The real pain points are:

• You cannot tell what actually changed between runs. You get the latest snapshot and nothing else.
• Late-arriving data is hard to reason about if your source keeps updating old rows.
• Downstream consumers (analytics, dashboards, ML models) want near-real-time data but your batch pipeline runs every 4 hours.

CDC solves these by reading the database write-ahead log (WAL) and emitting every insert, update, and delete as a separate event. You get a stream of changes rather than snapshots.

CDC Approaches at a Glance

There is more than one way to do CDC and the right choice depends on what you already have in your stack.

Approach	How It Works	Best When
Database triggers	Trigger on table writes to a changelog table	No external tools, small scale
Query-based	Track updated_at or an incrementing ID column	Simple, but misses deletes and can miss updates without proper design
WAL-based (Debezium)	Read the database transaction log directly	You need every change, including deletes, with low latency
Cloud-native replication	AWS DMS, GCP Datastream, etc.	You want a managed service and are okay with the cost

For this post we will focus on the WAL-based approach using Debezium because it is open-source, battle-tested, and gives you full control over what gets captured and where it goes.

What You Need

Before we set anything up, here is what we are working with:

• A PostgreSQL database (I am using version 15)
• Apache Kafka and Kafka Connect running somewhere (local Docker is fine for testing)
• A GCS bucket where we will land the CDC output in Parquet format

The architecture looks like this:

PostgreSQL WAL → Debezium Source Connector → Kafka Topic → S3/GCS Sink Connector → GCS (Parquet)

Step 1: Enable Logical Replication on PostgreSQL

Debezium reads the WAL, but PostgreSQL needs logical replication enabled for that. Check your postgresql.conf:

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

After changing these, restart PostgreSQL. You can check if it took with:

SHOW wal_level;

You also need to create a replication user. A dedicated user is safer than using the postgres superuser.

CREATE ROLE debezium_user WITH LOGIN PASSWORD 'a-strong-password' REPLICATION;
GRANT CONNECT ON DATABASE your_db TO debezium_user;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;

The REPLICATION privilege is what lets the user read the WAL. Without it, the connector will fail with an authentication error that is easy to misdiagnose if you do not know what you are looking for.

One thing I missed the first time around: if you create the replication slot manually using pg_create_logical_replication_slot, Debezium will not pick it up unless you point it at that slot explicitly. Let Debezium create its own slot. It handles the lifecycle better.

Step 2: Spin Up Kafka Connect with Debezium

If you are just testing locally, Docker Compose gets you up and running quickly. Here is a minimal set of services:

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

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

Once the connect container is up, you can check its health:

curl http://localhost:8083/connectors

Step 3: Configure the Debezium Source Connector

Now we tell Debezium which database to watch and what to capture.

{
  "name": "postgres-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "your-pg-host",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "a-strong-password",
    "database.dbname": "your_db",
    "topic.prefix": "cdc",
    "table.include.list": "public.orders,public.customers",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.autocreate.mode": "filtered",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false"
  }
}

A few things worth pointing out in that config:

• table.include.list is where you whitelist tables. Start small. Capturing every table in a busy database will flood your Kafka cluster before you know it.
• plugin.name should be pgoutput for PostgreSQL 10+. The older decoderbufs plugin works but needs an extra extension installed.
• I set key.converter.schemas.enable to false. If you are landing into a data lake, the Avro schema registry adds complexity you might not need yet. Plain JSON is easier to debug.

Submit this to the connect REST API:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @debezium-config.json

At this point Debezium takes a snapshot of the existing tables and then switches to streaming mode. You can see the topics it creates:

kafka-topics --list --bootstrap-server localhost:9092
# You should see: cdc.public.orders, cdc.public.customers

Step 4: Land the Data into GCS

Now that changes are flowing into Kafka, we need to get them somewhere useful. The S3 Sink connector works for GCS too since GCS has an S3-compatible API.

{
  "name": "gcs-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "cdc.public.orders,cdc.public.customers",
    "s3.bucket.name": "your-cdc-bucket",
    "s3.region": "us-central1",
    "s3.endpoint": "https://storage.googleapis.com",
    "store.url": "https://storage.googleapis.com",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "partitioner.class": "io.confluent.connect.storage.partitioner.DailyPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd",
    "partition.duration.ms": "86400000",
    "locale": "en-US",
    "timezone": "UTC"
  }
}

This writes Parquet files partitioned by date. Each Kafka message becomes a row, and the connector batches them up until it hits 10,000 records or 10 minutes, whichever comes first.

Things That Will Bother You in Practice

Replication slot bloat. If your Kafka consumer is down for a while, PostgreSQL keeps all the WAL segments referenced by the slot. Your disk will fill up and the database will stop accepting writes. Set up monitoring on replication slot lag — pg_stat_replication and pg_replication_slots are your friends here.

Schema changes. Debezium handles column additions fairly well, but dropping a column or changing a type will break things. You need a plan for schema evolution. In a simple setup, you just reprocess. At scale, you will want to think about Avro with a schema registry.

Delete events. Debezium emits delete events with only the primary key by default. If your downstream needs the full row that was deleted, set tombstones.on.delete to false and make sure REPLICA IDENTITY on your table is set to FULL. But be careful — REPLICA IDENTITY FULL writes the entire old row into the WAL, which increases WAL volume significantly.

Initial snapshot. The initial snapshot can take hours on a large table and locks the table while it runs. For production, you can set snapshot.mode to exported to use a less invasive snapshot, or use never if you already have the data loaded from another source.

What Changes for Production

This setup works for a proof of concept, but here is what I would change before calling it production-ready:

• Use a managed Kafka service (Confluent Cloud, MSK, etc.) instead of a single-node Docker Kafka. A broker going down should not kill your pipeline.
• Enable Kafka Connect’s distributed mode with at least two workers. The REST API we used above is single-node only.
• Switch from JSON to Avro with a Confluent Schema Registry. JSON is fine for debugging but Avro is smaller on the wire and gives you schema compatibility checks.
• Add dead-letter queuing for messages the sink connector cannot process. Without it, a single bad record can block the entire topic partition.
• Use IAM-based auth (Workload Identity in GCP, IRSA in AWS) instead of static access keys for the sink connector.
• Monitor connector lag using Kafka Connect’s JMX metrics. If the sink connector falls behind Kafka retention, you will lose data.

Wrapping Up

CDC is one of those things that feels complicated until you have set it up once. The Debezium plus Kafka Connect combo works well enough that I keep coming back to it even when there are managed alternatives available. The key is starting simple — one connector, a few tables, plain JSON output — and adding complexity only when you actually need it.

Once the CDC stream is landing in your data lake, you can build incremental ETL on top of it without ever scanning the source database again. That alone usually justifies the setup effort.