Distributed CDC Pipeline — Exactly-Once Delivery

Production-grade Change Data Capture (CDC) pipeline that guarantees exactly-once delivery from PostgreSQL to Apache Iceberg, surviving network partitions and Flink TaskManager crashes.

Built for 50,000 writes/minute upstream throughput with per-entity ordering, checkpoint-based recovery, and mathematical idempotency guarantees.

About

This project implements an end-to-end Change Data Capture (CDC) pipeline designed for production workloads where duplicate or lost events are unacceptable. PostgreSQL WAL changes are captured by Debezium, streamed through Kafka for durability and ordering, processed by a stateful Flink job with RocksDB-backed deduplication, and written to Apache Iceberg with two-phase commit semantics.

Key capabilities:

• Exactly-once delivery — Flink checkpoint barriers + Iceberg atomic commits + (entity_id, lsn, operation_type) idempotency keys
• Fault tolerance — survives TaskManager crashes and network partitions; recovers from the last completed checkpoint
• Per-entity ordering — Kafka partitioning by entity_id preserves event sequence per record
• Chaos-tested — includes a TaskManager kill script with DuckDB-based verification
• Docker-native — full stack (Postgres, Kafka, Flink, MinIO, Debezium) runs via docker compose with a single make start

Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                         EXACTLY-ONCE CDC PIPELINE                           │
└─────────────────────────────────────────────────────────────────────────────┘

  PostgreSQL 15                Kafka 3.5 (3 brokers)           Apache Flink 1.18
  wal_level=logical            12 partitions by entity_id      RocksDB + 30s checkpoints
  replication slot             read_committed isolation         KeyedProcessFunction dedup
       │                              │                                │
       │  WAL (pgoutput)              │  cdc.public.transactions         │
       ▼                              ▼                                ▼
 ┌───────────┐   Debezium Connect   ┌────────┐   PyFlink Job        ┌──────────┐
 │transactions│ ──────────────────► │ Kafka  │ ───────────────────► │ Dedup +  │
 │  table    │   flatten envelope   │ Topic  │   EXACTLY_ONCE       │ Iceberg  │
 └───────────┘                      └────────┘   source               │  Sink    │
                                                                      └────┬─────┘
                                                                           │
                                                                           ▼
                                                                    ┌──────────────┐
                                                                    │ MinIO (S3)   │
                                                                    │ checkpoints/ │
                                                                    │ iceberg-wh/  │
                                                                    └──────────────┘

Components

Layer	Technology	Role
Source	PostgreSQL 15	OLTP writes, logical replication slot
CDC	Debezium 2.5	WAL capture, envelope flattening
Broker	Kafka 3.5 × 3	Durable event log, per-entity ordering
Processor	Flink 1.18 PyFlink	Stateful dedup, exactly-once checkpoints
Sink	Iceberg v2 (warehouse on ./warehouse, MinIO-ready)	ACID upsert warehouse table
Chaos	flink_killer.py	TaskManager kill + DuckDB verification

Mathematical Proof of Exactly-Once

1. Checkpoint Barriers (Flink)

Flink injects checkpoint barriers into the data stream. When a barrier reaches an operator:

1. Operator snapshots its state (RocksDB) to s3a://checkpoints/flink
2. Operator acknowledges the barrier downstream
3. JobManager collects all acknowledgments → checkpoint complete

On failure, Flink restores the last completed checkpoint and replays Kafka from committed offsets. Uncommitted records from the failed epoch are never written to the sink.

2. Two-Phase Commit (Flink ↔ Iceberg)

With CheckpointingMode.EXACTLY_ONCE:

• Pre-commit: Iceberg writers stage data files but do not publish snapshots
• Checkpoint complete: Flink notifies sink → Iceberg commits snapshot atomically
• Abort on failure: Uncommitted files are discarded; no partial snapshots visible

Readers always see a consistent Iceberg snapshot — never a half-written batch.

3. Idempotency Key: (entity_id, lsn, operation_type)

PostgreSQL WAL assigns a monotonically increasing LSN per change. Our KeyedProcessFunction:

key   = entity_id
state = MapState[(entity_id|lsn|op) → seen] + ValueState[last_lsn]

if seen(entity_id, lsn, op):  SKIP   # exact duplicate on replay
if lsn <= last_lsn:           SKIP   # out-of-order duplicate
else:                         EMIT + update state

Why this is sufficient:

• Each WAL event has a unique LSN globally
• Replays (from checkpoint recovery or Kafka redelivery) present the same (entity_id, lsn, op) tuple
• State TTL (24h) bounds memory while covering any realistic recovery window
• Iceberg upsert on entity_id provides a final convergence guarantee

Composition theorem:

exactly_once(Flink) × idempotent(dedup) × atomic_commit(Iceberg) = exactly_once(end-to-end)

Duplicate delivery attempts are absorbed by dedup state; the Iceberg sink never exposes uncommitted or duplicated snapshots to readers.

Quick Start

Prerequisites

• Docker Desktop 4.x+ with Compose v2
• Python 3.10+
• 8 GB RAM available for containers

1. Install Python dependencies

pip install -r requirements.txt

2. Start infrastructure

make start

This builds Flink images, starts all services, and waits for health checks.

3. Deploy Debezium connector

make deploy-connector

4. Submit Flink pipeline

make run-pipeline

5. Run chaos test

make run-chaos

6. Run full test suite

make test

Service Endpoints

Service	URL
Kafka UI	http://localhost:8080
Flink Dashboard	http://localhost:8081
Kafka Connect	http://localhost:8083
MinIO Console	http://localhost:9001
PostgreSQL	localhost:5432

Project Structure

├── docker-compose.yml          # Full stack orchestration
├── Makefile                    # start, deploy-connector, run-pipeline, run-chaos, test
├── postgres/init.sql           # Schema, publication, seed data
├── debezium/postgres-connector.json
├── flink/                      # Custom Flink image + S3/RocksDB config
├── flink_jobs/cdc_processor.py # PyFlink exactly-once pipeline
├── producer/load_generator.py  # 1000-event test load
├── chaos/flink_killer.py       # TaskManager kill + verification
├── verification/check_exactly_once.py
└── tests/test_exactly_once.py

Performance Characteristics

Metric	Value
End-to-end latency p50	29.3s
End-to-end latency p95	31.4s
End-to-end latency p99	31.5s
Sustained throughput	167 records/s
Checkpoint size (latest)	0.51 MB
Checkpoint duration (latest)	281 ms
Consumer group lag	10,000 messages

Measured on Docker Desktop (local dev). Run POSTGRES_PORT=5433 py -3.11 scripts/benchmark.py to regenerate.

Makefile Targets

Target	Description
make start	Build and start all containers, wait for health
make deploy-connector	Register Debezium PostgreSQL connector
make run-pipeline	Deploy connector + submit Flink job
make run-chaos	Insert 1000 events, kill TaskManager, verify
make test	Run pytest integration suite
make stop	Stop containers
make clean	Stop and remove volumes

Troubleshooting

Connector fails with "publication does not exist"

• Ensure postgres/init.sql ran: docker exec cdc-postgres psql -U cdc_user -d cdc_db -c "\dRp"

Flink job cannot reach MinIO

• Verify buckets: docker run --rm --network distributedcdc-pipeline_cdc-net minio/mc ls local/
• Check s3.endpoint in flink/conf/flink-conf.yaml

Chaos test times out

• Confirm Flink job is RUNNING at http://localhost:8081
• Increase wait in chaos/flink_killer.py if hardware is slow

License

MIT