Case 2 — Real-Time Ingestion: Postgres CDC to Cloud Warehouse

# Case 2 — Real-Time Ingestion: Postgres CDC to Cloud Warehouse ## Executive Context & Business Prompt You are the Lead Pipeline Engineer at **FinFlow Technologies**, a fast-growing neobank. The core banking application runs on PostgreSQL, recording customer ledger accounts, debit card authorizations, and merchant transfers. Analytics dashboards and anti-money-laundering (AML) models currently query a secondary Postgres read-replica via hourly batch cron jobs. This setup is breaking under load: 1. **Replica Query Contention:** Analytical queries locking tables cause replication lag to spike from 2 seconds to 35 minutes, violating banking regulator freshness SLAs. 2. **Missing Intermediate Transitions:** A payment record transitioning from `pending` -> `authorized` -> `settled` within a 10-minute window only appears as `settled` in the hourly extract, blinding the AML model to rapid velocity fraud. 3. **Hard Deletes Disappear:** Regulatory compliance requires archiving closed account records, but physical `DELETE` statements on Postgres leave zero trace in the hourly batch delta. Head of Data Prompt: > *"We cannot query production OLTP tables directly anymore. Stand up a zero-impact Change Data Capture (CDC) pipeline using Postgres logical replication (WAL) and Debezium. Stream every insert, update (before & after values), and delete into Kafka/S3 with sub-10-second lag."* --- ## Explicit Portfolio Deliverables 1. **Infrastructure Scaffold (`docker-compose.yml`):** - Configures PostgreSQL 15+ with `wal_level=logical`, `max_wal_senders=10`, and `max_replication_slots=10`. - Provisions Apache Kafka and Debezium Connect with health checks and network bridging. 2. **Debezium Connector Configuration (`postgres-cdc-connector.json`):** - Connects to Postgres using `pgoutput` plugin. - Configures table inclusion whitelist: `public.accounts`, `public.transactions`. - Enables tombstone records and schema evolution capture. 3. **Replication Slot Monitoring & SLA Query (`slot_monitor.sql`):** - Author SQL monitoring query against `pg_replication_slots` and `pg_stat_replication` calculating WAL lag in megabytes. - Raises an alert when lag exceeds 500MB to prevent primary node disk exhaustion. --- ## Starter Code Scaffold ```yaml version: '3.8' services: postgres: image: postgres:15 environment: POSTGRES_DB: banking POSTGRES_USER: postgres POSTGRES_PASSWORD: secretpassword command: ["postgres", "-c", "wal_level=logical", "-c", "max_replication_slots=5"] ports: - "5432:5432" kafka: image: confluentinc/cp-kafka:7.4.0 ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT' KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:9092' KAFKA_PROCESS_ROLES: 'broker,controller' KAFKA_CONTROLLER_QUORUM_VOTERS: '1@localhost:9093' KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093' KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' debezium: image: debezium/connect:2.4 ports: - "8083:8083" environment: BOOTSTRAP_SERVERS: kafka:9092 GROUP_ID: 1 CONFIG_STORAGE_TOPIC: my_connect_configs OFFSET_STORAGE_TOPIC: my_connect_offsets STATUS_STORAGE_TOPIC: my_connect_statuses ``` --- ## 100-Point Grading Rubric | Criteria | Points | Passing Standard | |---|:---:|---| | **1. Postgres WAL & Slot Configuration** | 25 pts | Correctly configures PostgreSQL logical replication settings (`wal_level=logical`) and creates dedicated replication slot without database restarts. | | **2. Debezium Connector Spec & Routing** | 25 pts | Valid JSON connector payload using `pgoutput`, correctly whitelisting target tables, specifying snapshot modes, and configuring Kafka topic prefixes. | | **3. Change Payload Capture (Before/After/Deletes)** | 25 pts | Emits complete change events capturing `before` state on updates, `after` state on inserts, and `op: 'd'` tombstones for physical deletes. | | **4. Operational Monitoring & Lag Prevention** | 25 pts | Author diagnostic query against `pg_replication_slots` computing byte lag to safeguard primary disk storage against unconsumed slot growth. |