CDC Pipeline: PostgreSQL to Snowflake

Introduction

30 Second Summary

Every time you place an order online, dozens of downstream systems need to know about it instantly. The analytics dashboard updates, the warehouse gets notified, and fraud detection kicks in. This real-time flow from database to everywhere else is called Change Data Capture, and it powers the streaming infrastructure at companies like Netflix and Uber.

In this project, you will build a production-grade CDC pipeline that captures every INSERT, UPDATE, and DELETE from a PostgreSQL e-commerce database, streams it through Kafka with schema governance, processes it in real-time with Apache Flink, and sinks the enriched data into Snowflake for cloud-based analytics.

What You'll Build

A fully containerized real-time data pipeline using Docker Compose to orchestrate 9 services that capture database changes, serialize them with Avro, aggregate them with Flink SQL, and deliver them to a cloud data warehouse.

By the end of this project, you'll have:

• A Debezium CDC connector reading the PostgreSQL Write-Ahead Log and producing Avro-serialized change events to Kafka with Schema Registry governance.
• A Flink SQL streaming job performing real-time revenue aggregation by product category, writing enriched results to a downstream Kafka topic.
• A Snowflake sink connector delivering processed data to a cloud warehouse using key-pair authentication and Snowpipe Streaming for low-latency ingestion.
• A Prometheus + Grafana observability stack monitoring consumer lag, connector status, and pipeline throughput.
• Secret Mission: add Dead Letter Queue error handling, enable exactly-once delivery semantics, and inject malformed records to prove the pipeline handles failures gracefully.

Are there any prerequisites?

This project assumes familiarity with Docker, Kafka, and SQL. You should be comfortable running Docker Compose commands and writing basic SQL queries.

You will need Docker Desktop installed with at least 8 GB of memory allocated, Python 3 for the data generator script, and OpenSSL for key generation. The Snowflake free trial provides $400 in credits for 30 days with no credit card required.

Set Up Project Structure and Snowflake

A real-time CDC pipeline has many moving parts. You need separate configurations for your database, message broker, connectors, stream processor, and monitoring stack.

In this step, you will scaffold the project directory, create a Snowflake free trial account with the required database resources, and generate an RSA key pair for secure authentication between Kafka Connect and Snowflake.

In this step, get ready to:

• Create the project folder structure with dedicated directories for each service.
• Set up a Snowflake free trial and configure database resources.
• Generate an RSA key pair for secure connector authentication.

Create the project directory structure

Each service in the pipeline needs its own configuration files. Organizing them into separate directories now keeps the project manageable as it grows.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and all subdirectories by running these commands:

mkdir -p cdc-pipeline/{connect,flink,postgres,connectors,monitoring/grafana/provisioning,scripts,secrets}
cd cdc-pipeline

What do these directories hold?

• connect/ holds the custom Kafka Connect Dockerfile with Debezium and Snowflake plugins.
• flink/ holds the custom Flink Dockerfile with Kafka and Avro connectors.
• postgres/ holds the database initialization script.
• connectors/ holds JSON configurations for Debezium and Snowflake connectors.
• monitoring/ holds Prometheus and Grafana configuration files.
• scripts/ holds utility scripts like the data generator and Snowflake setup SQL.
• secrets/ holds authentication keys that Docker will mount into containers.

• Confirm the directories were created by running:

ls

You should see connect, connectors, flink, monitoring, postgres, scripts, and secrets listed in the output.

Don't see all the folders?

Make sure you included the curly braces {...} in the mkdir command. Brace expansion is a shell feature that creates multiple directories in one go.

If you are on Windows using PowerShell, brace expansion is not supported. Create each folder individually instead: mkdir connect, flink, postgres, connectors, monitoring/grafana/provisioning, scripts, secrets.

help me create the project directories

Set up your Snowflake account and resources

Snowflake is the cloud data warehouse where your pipeline's enriched data will land. You need a free trial account and a few database objects configured before the connector can write data.

• Open your browser and navigate to signup.snowflake.com.
• Enter your name and email address.
• Select Personal learning and development for the signup reason.
• Click Continue.
• On the next screen, select Enterprise as your Snowflake edition.
• Choose the cloud provider and region closest to you.
• Click Get Started.

What is the Snowflake free trial?

The trial gives you 30 days and $400 in free credits with no credit card required. You will receive an activation email with a link to access your account.

The X-Small warehouse you will create auto-suspends after 5 minutes of inactivity, so it will not burn through credits while idle.

• Check your email for the activation link and click it to set your password.
• Log in to your Snowflake account. You will land in the Snowsight web interface.

Now you will create the database resources that the Kafka connector needs. First, save a reference SQL script in your project.

• Create the setup script by running this command from your cdc-pipeline folder:

touch scripts/setup_snowflake.sql

• Open scripts/setup_snowflake.sql in your text editor and paste the following SQL:

-- Run these commands in the Snowflake web UI (Snowsight)

-- Create dedicated database and schema
CREATE DATABASE IF NOT EXISTS CDC_PIPELINE;
USE DATABASE CDC_PIPELINE;
CREATE SCHEMA IF NOT EXISTS REALTIME;

-- Create a warehouse for the connector
CREATE WAREHOUSE IF NOT EXISTS CDC_WAREHOUSE
  WAREHOUSE_SIZE = 'X-SMALL'
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE;

-- Create a dedicated role and user for the Kafka connector
USE ROLE SECURITYADMIN;
CREATE ROLE IF NOT EXISTS CDC_KAFKA_ROLE;
CREATE USER IF NOT EXISTS CDC_KAFKA_USER
  DEFAULT_ROLE = CDC_KAFKA_ROLE
  DEFAULT_WAREHOUSE = CDC_WAREHOUSE;
GRANT ROLE CDC_KAFKA_ROLE TO USER CDC_KAFKA_USER;

-- Grant privileges
USE ROLE SYSADMIN;
GRANT USAGE ON DATABASE CDC_PIPELINE TO ROLE CDC_KAFKA_ROLE;
GRANT USAGE ON SCHEMA CDC_PIPELINE.REALTIME TO ROLE CDC_KAFKA_ROLE;
GRANT CREATE TABLE ON SCHEMA CDC_PIPELINE.REALTIME TO ROLE CDC_KAFKA_ROLE;
GRANT CREATE STAGE ON SCHEMA CDC_PIPELINE.REALTIME TO ROLE CDC_KAFKA_ROLE;
GRANT CREATE PIPE ON SCHEMA CDC_PIPELINE.REALTIME TO ROLE CDC_KAFKA_ROLE;
GRANT USAGE ON WAREHOUSE CDC_WAREHOUSE TO ROLE CDC_KAFKA_ROLE;

What does this SQL do?

• CDC_PIPELINE is the database where enriched pipeline data will land.
• REALTIME is the schema inside that database for streaming tables.
• CDC_WAREHOUSE is an X-Small compute warehouse that auto-suspends after 5 minutes to save credits.
• CDC_KAFKA_ROLE and CDC_KAFKA_USER isolate the connector's permissions from your admin account.
• The GRANT statements give the role only the minimum privileges needed: creating tables, stages, and pipes.

• In Snowsight, click Worksheets in the left sidebar.
• Click the + Worksheet button to create a new SQL worksheet.
• Copy the entire contents of scripts/setup_snowflake.sql and paste it into the worksheet.
• Select all the SQL and click Run (or press Ctrl+Enter).

Each statement should return a success message. You can verify by expanding CDC_PIPELINE in the left database browser and confirming the REALTIME schema appears.

Seeing a permission error?

Make sure you are running the worksheet as ACCOUNTADMIN. Check the role selector in the top-right of the worksheet and switch to ACCOUNTADMIN if needed.

help me fix a permission error when running setup SQL in Snowsight

Generate an RSA key pair for authentication

Production systems authenticate connectors using key pairs instead of passwords. The Snowflake Kafka Connector will use an RSA private key to prove its identity without storing a password in your config files.

Why key-pair authentication instead of a password?

Passwords in configuration files are a security risk. Key-pair authentication uses cryptography: you give Snowflake the public key, and the connector proves its identity with the matching private key.

This is the standard approach for production Kafka-to-Snowflake pipelines. The Snowflake Kafka Connector v3.5.2 fat JAR supports unencrypted private keys, so no passphrase is needed.

• Generate an unencrypted RSA private key in PKCS#8 format by running this command from your cdc-pipeline folder:

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out secrets/snowflake_rsa_key.p8 -nocrypt

What does this command do?

• openssl genrsa 2048 generates a 2048-bit RSA key (the minimum Snowflake requires).
• openssl pkcs8 -topk8 converts the key to PKCS#8 format, which Snowflake requires.
• -nocrypt skips passphrase encryption, which is required by the Snowflake Kafka Connector fat JAR.

• Extract the corresponding public key by running:

openssl rsa -in secrets/snowflake_rsa_key.p8 -pubout -out secrets/snowflake_rsa_key.pub

• Verify both keys exist by running:

ls secrets/

You should see snowflake_rsa_key.p8 and snowflake_rsa_key.pub in the output.

Getting a 'command not found' error?

Make sure OpenSSL is installed. On macOS it comes pre-installed. On Windows, install it from the OpenSSL for Windows binaries or use Git Bash which bundles OpenSSL.

help me install or find OpenSSL on my system

Now you need to tell Snowflake about the public key so it can verify the connector's identity.

• Display the public key body (without the header and footer lines) by running:

grep -v "BEGIN\|END" secrets/snowflake_rsa_key.pub | tr -d '\n'

This outputs a single line of base64 text. Copy the entire output to your clipboard.

• Go back to your Snowsight worksheet.
• Make sure the role is set to SECURITYADMIN in the role selector.
• Run the following command, replacing your-public-key-here with the base64 string you copied:

ALTER USER CDC_KAFKA_USER SET RSA_PUBLIC_KEY='your-public-key-here';

Paste the key as one continuous line

The public key must be a single unbroken string inside the quotes with no line breaks or spaces. The grep and tr command above already strips the headers and joins the lines for you.

• Verify the key was assigned by running this in Snowsight:

DESC USER CDC_KAFKA_USER;

You should see a row with RSA_PUBLIC_KEY_FP containing a SHA-256 fingerprint value. This confirms Snowflake recognized the key.

Finally, create the private key file in the format that Kafka Connect's file config provider expects.

• Create the connector-ready private key file by running:

echo "private_key=$(grep -v 'BEGIN\|END' secrets/snowflake_rsa_key.p8 | tr -d '\n')" > secrets/snowflake-private-key.txt

Why a separate private key file?

The Snowflake connector config uses Kafka Connect's FileConfigProvider to read secrets from files at runtime. It expects a Java properties format: private_key=<key-body>.

This keeps the actual key out of your connector JSON config, which will be committed to version control.

• Confirm the file was created by running:

ls secrets/

You should now see three files: snowflake_rsa_key.p8, snowflake_rsa_key.pub, and snowflake-private-key.txt.

DESC USER shows no RSA fingerprint?

Double-check that you ran the ALTER USER command as SECURITYADMIN. Also verify the key string has no line breaks or spaces inside the quotes.

If the RSA_PUBLIC_KEY_FP row is still empty, try re-running the grep command and carefully re-pasting the output.

help me assign my RSA public key to CDC_KAFKA_USER in Snowflake

Your project structure is ready and Snowflake is configured to accept connections from the Kafka connector. Next up, you will build the Docker infrastructure that runs Kafka, Schema Registry, PostgreSQL, and Kafka Connect.

Build the Docker Infrastructure

Your project structure and Snowflake account are ready. Now it's time to build the core infrastructure that makes real-time CDC possible.

Every service in your pipeline needs to communicate reliably. Kafka brokers messages between producers and consumers. Schema Registry governs the shape of those messages. PostgreSQL stores the source data. And Kafka Connect bridges them all together. In this step, you will define all four services in a single Docker Compose file and bring the stack to life.

In this step, get ready to:

• Create a Docker Compose file with Kafka (KRaft mode), Schema Registry, PostgreSQL, and Kafka Connect.
• Build a custom Kafka Connect image with Debezium, Avro, and the Snowflake connector.
• Start the stack and verify every service is healthy.

Create the Docker Compose file

Docker Compose lets you define and run multi-container applications with a single YAML file. Your pipeline needs four services that depend on each other in a specific order.

• Create a file called docker-compose.yml in your cdc-pipeline project root.
• Add the Kafka service definition by pasting this code:

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.8
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092"
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:29093"
      KAFKA_LISTENERS: "PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092"
      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LOG_DIRS: "/tmp/kraft-combined-logs"
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
    healthcheck:
      test: kafka-topics --bootstrap-server kafka:29092 --list
      interval: 10s
      timeout: 5s
      retries: 5

What is KRaft mode?

Traditional Kafka deployments require a separate ZooKeeper cluster to manage metadata. KRaft (Kafka Raft) eliminates that dependency by embedding consensus directly into the broker.

The key settings making this work are KAFKA_PROCESS_ROLES: "broker,controller" (combined mode for local development) and CLUSTER_ID (a fixed ID so the cluster initialises deterministically). ZooKeeper is deprecated for new deployments as of Confluent Platform 7.5.

• Below the kafka service in the same file, add the Schema Registry and PostgreSQL services:

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.8
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "kafka:29092"
      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"

  postgres:
    image: postgres:16
    hostname: postgres
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: cdc_user
      POSTGRES_PASSWORD: cdc_password
      POSTGRES_DB: ecommerce
    command: >
      postgres
      -c wal_level=logical
      -c max_wal_senders=4
      -c max_replication_slots=4
    volumes:
      - ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql

Why is wal_level set to logical?

PostgreSQL's Write-Ahead Log (WAL) records every change before it hits disk. Setting wal_level=logical enables logical replication, which is what Debezium reads to capture individual row-level changes.

The max_wal_senders and max_replication_slots settings control how many replication connections PostgreSQL allows. Four is more than enough for local development.

• Below the postgres service, add the Kafka Connect service:

  kafka-connect:
    build:
      context: ./connect
      dockerfile: Dockerfile
    hostname: kafka-connect
    container_name: kafka-connect
    depends_on:
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_started
      postgres:
        condition: service_started
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka:29092"
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
      CONNECT_GROUP_ID: cdc-connect-cluster
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
    volumes:
      - ./secrets:/etc/kafka-connect/secrets

What does this service do?

Kafka Connect is the framework that runs your source (Debezium) and sink (Snowflake) connectors. It builds from a custom Dockerfile you will create next.

The AvroConverter settings tell Connect to serialize all messages using Avro and register schemas with Schema Registry. This gives you schema governance across the entire pipeline.

The secrets volume mount exposes your Snowflake private key to the connector without baking credentials into the image.

• Save docker-compose.yml.

YAML indentation errors?

YAML is whitespace-sensitive. Every service must be indented exactly 2 spaces under services:. Environment variables are indented 6 spaces (2 for service, 2 for environment:, 2 for the key). If your editor uses tabs instead of spaces, Docker Compose will fail to parse the file.

help me fix YAML indentation in my docker-compose.yml

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml before continuing.

ⓧ I'd like to double check the full code

Here is the full docker-compose.yml you should have at this point:

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.8
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092"
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:29093"
      KAFKA_LISTENERS: "PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092"
      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LOG_DIRS: "/tmp/kraft-combined-logs"
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
    healthcheck:
      test: kafka-topics --bootstrap-server kafka:29092 --list
      interval: 10s
      timeout: 5s
      retries: 5

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.8
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "kafka:29092"
      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"

  postgres:
    image: postgres:16
    hostname: postgres
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: cdc_user
      POSTGRES_PASSWORD: cdc_password
      POSTGRES_DB: ecommerce
    command: >
      postgres
      -c wal_level=logical
      -c max_wal_senders=4
      -c max_replication_slots=4
    volumes:
      - ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql

  kafka-connect:
    build:
      context: ./connect
      dockerfile: Dockerfile
    hostname: kafka-connect
    container_name: kafka-connect
    depends_on:
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_started
      postgres:
        condition: service_started
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka:29092"
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
      CONNECT_GROUP_ID: cdc-connect-cluster
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
    volumes:
      - ./secrets:/etc/kafka-connect/secrets

The Docker Compose file references an init.sql file that PostgreSQL will execute automatically when it first starts. You need to create this file before bringing the stack up.

• Create a file called init.sql inside your postgres/ folder with this content:

-- Enable logical replication (already set via command line, but confirm)
-- Create the e-commerce schema

CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(150) UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,
    name VARCHAR(200) NOT NULL,
    category VARCHAR(50) NOT NULL,
    price DECIMAL(10, 2) NOT NULL
);

CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(customer_id),
    product_id INTEGER REFERENCES products(product_id),
    quantity INTEGER NOT NULL DEFAULT 1,
    total_amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Set REPLICA IDENTITY FULL on all tables
-- Required for Flink to interpret Debezium PostgreSQL CDC data
ALTER TABLE customers REPLICA IDENTITY FULL;
ALTER TABLE products REPLICA IDENTITY FULL;
ALTER TABLE orders REPLICA IDENTITY FULL;

-- Insert seed data
INSERT INTO customers (name, email) VALUES
    ('Alice Johnson', 'alice@example.com'),
    ('Bob Smith', 'bob@example.com'),
    ('Carol Williams', 'carol@example.com');

INSERT INTO products (name, category, price) VALUES
    ('Wireless Headphones', 'Electronics', 79.99),
    ('Running Shoes', 'Sports', 129.99),
    ('Coffee Maker', 'Home', 49.99),
    ('Laptop Stand', 'Electronics', 39.99),
    ('Yoga Mat', 'Sports', 29.99);

INSERT INTO orders (customer_id, product_id, quantity, total_amount, status) VALUES
    (1, 1, 1, 79.99, 'completed'),
    (2, 3, 2, 99.98, 'completed'),
    (1, 2, 1, 129.99, 'pending');

Why REPLICA IDENTITY FULL?

By default, PostgreSQL only logs the primary key and changed columns in UPDATE events. This is a problem for Flink SQL, which needs the full before-and-after state of every row to process CDC correctly.

Setting REPLICA IDENTITY FULL on each table tells PostgreSQL to log all column values on every change. Without this, your Flink streaming queries will fail later in the project.

• Save postgres/init.sql.

Create the Kafka Connect Dockerfile

Kafka Connect needs three plugins that are not included in the base image: the Debezium CDC connector, the Avro converter for Schema Registry integration, and the Snowflake sink connector. You will build a custom image that bundles all three.

• Create a file called Dockerfile inside your connect/ folder with this content:

FROM confluentinc/cp-kafka-connect-base:7.7.8

# Install Debezium PostgreSQL connector
RUN confluent-hub install --no-prompt debezium/debezium-connector-postgresql:3.0.0

# Install Avro converter (for Schema Registry integration)
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-avro-converter:7.7.8

# Download Snowflake Kafka Connector fat JAR from Maven Central
RUN mkdir -p /usr/share/confluent-hub-components/snowflake-kafka-connector && \
    curl -L -o /usr/share/confluent-hub-components/snowflake-kafka-connector/snowflake-kafka-connector-3.5.2.jar \
    https://repo1.maven.org/maven2/com/snowflake/snowflake-kafka-connector/3.5.2/snowflake-kafka-connector-3.5.2.jar

What does each layer install?

• debezium/debezium-connector-postgresql:3.0.0 is the CDC source connector that reads the PostgreSQL WAL.
• confluentinc/kafka-connect-avro-converter:7.7.8 enables Avro serialization so every message is schema-governed.
• snowflake-kafka-connector-3.5.2.jar is the Snowflake sink connector. The fat JAR (199 MB) bundles all dependencies so no extra libraries are needed.

• Save connect/Dockerfile.

Worried about the large download?

The Snowflake connector JAR is approximately 199 MB. The download happens during the Docker image build, so a slow connection may cause the build to take several minutes. This is normal.

help me troubleshoot my Kafka Connect Dockerfile build failing

✔️ Awesome, I've got everything!

Great. Make sure you have saved connect/Dockerfile before continuing.

ⓧ I'd like to double check the full code

Here is the complete connect/Dockerfile:

FROM confluentinc/cp-kafka-connect-base:7.7.8

# Install Debezium PostgreSQL connector
RUN confluent-hub install --no-prompt debezium/debezium-connector-postgresql:3.0.0

# Install Avro converter (for Schema Registry integration)
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-avro-converter:7.7.8

# Download Snowflake Kafka Connector fat JAR from Maven Central
RUN mkdir -p /usr/share/confluent-hub-components/snowflake-kafka-connector && \
    curl -L -o /usr/share/confluent-hub-components/snowflake-kafka-connector/snowflake-kafka-connector-3.5.2.jar \
    https://repo1.maven.org/maven2/com/snowflake/snowflake-kafka-connector/3.5.2/snowflake-kafka-connector-3.5.2.jar

Start all services and verify health

With all configuration files in place, you can now bring the entire stack online. Docker Compose will build the custom Kafka Connect image, pull the official images, and start services in dependency order.

Allocate enough memory to Docker

Confluent recommends at least 8 GB of RAM allocated to Docker Desktop when running Kafka. Open Docker Desktop, go to Settings then Resources, and confirm the memory slider is set to 8 GB or higher.

• In your terminal, make sure you are inside the cdc-pipeline project folder.
• Start all services in detached mode by running:

docker compose up -d

What happens when you run this?

Docker Compose builds the custom Kafka Connect image (this takes a few minutes on first run), pulls the official Kafka, Schema Registry, and PostgreSQL images, then starts all four containers.

The depends_on conditions ensure services start in the right order. Kafka must be healthy before Schema Registry or Kafka Connect start. The Kafka healthcheck runs kafka-topics --list every 10 seconds to confirm the broker is ready.

• Wait 1-2 minutes for all services to start, then verify Schema Registry is reachable by running:

curl -s http://localhost:8081/subjects | head

You should see an empty JSON array [] indicating Schema Registry is running but no schemas are registered yet.

• Verify Kafka Connect is running and has no connectors yet:

curl -s http://localhost:8083/connectors

You should see [] again. The Connect framework is ready and waiting for connector configurations.

• Verify PostgreSQL is accepting connections by running:

docker exec postgres psql -U cdc_user -d ecommerce -c "\dt"

You should see a table listing showing customers, products, and orders. This confirms the init script ran successfully.

Services not starting?

Run docker compose ps to check container status. If kafka-connect shows as Restarting, it may still be waiting for Kafka's healthcheck to pass. Give it another minute.

If you see a build error related to the Snowflake JAR download, check your internet connection. The 199 MB file must download completely during the image build.

If PostgreSQL shows no tables, the init.sql file may not be in the correct path. Confirm it exists at postgres/init.sql relative to your docker-compose.yml.

help me debug my Docker Compose services not starting for my CDC pipeline

Your infrastructure is live. Kafka is brokering messages, Schema Registry is ready to govern schemas, PostgreSQL has your e-commerce data, and Kafka Connect is standing by. Next up, you will configure the Debezium CDC connector to start streaming database changes into Kafka.

Configure CDC Source and Verify Events

Your Docker infrastructure is running and healthy. Now it's time to activate the core mechanism that makes this pipeline real-time: Change Data Capture.

In this step, you'll register a Debezium connector that reads PostgreSQL's Write-Ahead Log (WAL) and publishes every INSERT, UPDATE, and DELETE as an Avro-serialized event to Kafka. You'll then verify that events flow through correctly and that schemas are governed by the Schema Registry.

In this step, get ready to:

• Create the Debezium connector configuration file with Avro serialization settings.
• Register the connector via the Kafka Connect REST API and confirm it is running.
• Verify CDC events appear in Kafka topics with auto-registered Avro schemas.

Create the Debezium connector configuration

Kafka Connect uses JSON configuration files to define connectors. Each file tells Connect which connector class to use, which database to connect to, and how to serialize the output.

• Create a connectors directory inside your project folder by running this command:

mkdir -p connectors

• Create a file called debezium-source.json inside the connectors folder and paste the following configuration:

{
  "name": "postgres-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "cdc_user",
    "database.password": "cdc_password",
    "database.dbname": "ecommerce",
    "topic.prefix": "cdc",
    "schema.include.list": "public",
    "table.include.list": "public.orders,public.customers,public.products",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "publication.autocreate.mode": "filtered",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

What does this configuration do?

This JSON tells Kafka Connect to launch a Debezium PostgreSQL connector with these key behaviors:

• topic.prefix: "cdc" means each table gets a topic named cdc.public.<table_name>.
• plugin.name: "pgoutput" uses the native PostgreSQL logical decoding output plugin, recommended for PostgreSQL 10+.
• snapshot.mode: "initial" takes a full snapshot of existing data on first startup, then switches to streaming WAL changes.
• The key.converter and value.converter settings serialize all messages as Avro and register schemas with Schema Registry automatically.

• Save connectors/debezium-source.json.
• Confirm the file exists by running:

ls connectors/

You should see debezium-source.json in the output.

Register the Debezium connector via the REST API

Kafka Connect exposes a REST API on port 8083 for managing connectors. You register a connector by POSTing its JSON config to the /connectors endpoint.

• Register the Debezium source connector by running this command:

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

What does this command do?

The curl -X POST sends your connector JSON to Kafka Connect's REST API. The -d @connectors/debezium-source.json flag reads the file contents as the request body. Kafka Connect then loads the Debezium plugin, connects to PostgreSQL, creates a replication slot, and begins streaming.

You should see a JSON response containing the full connector config echoed back. This confirms Kafka Connect accepted the registration.

• Wait about 10 seconds for the connector to initialize, then check its status by running:

curl -s http://localhost:8083/connectors/postgres-cdc-source/status | python3 -m json.tool

You should see "state": "RUNNING" in both the connector and task sections.

Connector showing FAILED?

• Check that the PostgreSQL container is running with docker ps. The connector cannot start if it cannot reach postgres:5432.
• If the error mentions "logical decoding" or "wal_level", confirm the PostgreSQL container started with wal_level=logical by running docker exec postgres psql -U cdc_user -d ecommerce -c "SHOW wal_level;".
• If you see a plugin error, verify the Debezium plugin is installed by running docker exec kafka-connect ls /usr/share/confluent-hub-components/.

My Debezium connector status shows FAILED. How do I debug this?

Verify CDC events and schema registration

When the connector started with snapshot.mode: "initial", it captured all existing rows in PostgreSQL and published them to Kafka. The seed data you inserted during setup should already be in the topics.

• List the Kafka topics to confirm the CDC topics were created by running:

docker exec kafka kafka-topics --bootstrap-server kafka:29092 --list

You should see cdc.public.orders, cdc.public.customers, and cdc.public.products in the list (along with internal Kafka Connect topics).

• Verify that Avro schemas were automatically registered in Schema Registry by running:

curl -s http://localhost:8081/subjects | python3 -m json.tool

You should see subjects like cdc.public.orders-key, cdc.public.orders-value, and matching key/value pairs for customers and products.

Why does Schema Registry matter here?

Every message Debezium produces is serialized as Avro and its schema is registered with Schema Registry. This means downstream consumers (Flink, Snowflake) can deserialize messages reliably without hardcoding schemas.

Schema Registry also enforces compatibility rules. If a schema changes in an incompatible way, it rejects the registration. This prevents breaking changes from crashing your pipeline.

Now let's generate a fresh CDC event and inspect its structure.

• Insert a new order into PostgreSQL by running:

docker exec postgres psql -U cdc_user -d ecommerce -c "
INSERT INTO orders (customer_id, product_id, quantity, total_amount, status)
VALUES (3, 4, 2, 79.98, 'pending');
"

• Consume the latest event from the cdc.public.orders topic to see the CDC event structure by running:

docker exec kafka kafka-console-consumer \
  --bootstrap-server kafka:29092 \
  --topic cdc.public.orders \
  --from-beginning \
  --max-messages 1 \
  --property print.key=true

What should I see?

The output will look garbled because it's Avro-encoded binary, not plain JSON. This is expected. The key thing is that a message appears at all, confirming CDC is working.

Debezium events follow an envelope structure with before (the row state before the change) and after (the row state after the change) fields. For INSERTs, before is null. For DELETEs, after is null.

• To see a human-readable version of the schema structure, inspect the registered value schema by running:

curl -s http://localhost:8081/subjects/cdc.public.orders-value/versions/latest | python3 -m json.tool

The response shows the Avro schema with the full Debezium envelope. You should see fields like before, after, source, and op (operation type: c=create, u=update, d=delete).

No messages appearing in the topic?

• Confirm the connector status is still RUNNING with curl -s http://localhost:8083/connectors/postgres-cdc-source/status.
• Check Kafka Connect logs for errors with docker logs kafka-connect --tail 50.
• Verify the INSERT succeeded by running docker exec postgres psql -U cdc_user -d ecommerce -c "SELECT * FROM orders;".

I registered my Debezium connector but no CDC events appear in Kafka. How do I troubleshoot?

✔️ Awesome, I've got everything!

Great. Double check you've saved connectors/debezium-source.json and confirmed your connector is RUNNING.

ⓧ I'd like to double check the full code

Here is the complete connector configuration file:

{
  "name": "postgres-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "cdc_user",
    "database.password": "cdc_password",
    "database.dbname": "ecommerce",
    "topic.prefix": "cdc",
    "schema.include.list": "public",
    "table.include.list": "public.orders,public.customers,public.products",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "publication.autocreate.mode": "filtered",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Your CDC source is live. Every change to the PostgreSQL database now streams through Kafka as an Avro-encoded event with full schema governance. Next up, you'll build a Flink SQL job that consumes these events and performs real-time aggregations.

Stream Processing with Flink SQL

CDC events are flowing from PostgreSQL through Debezium into Kafka topics. But raw change events are not directly useful for analytics.

In this step, you will deploy Apache Flink as a stream processing engine to continuously aggregate revenue by product category. Flink reads the Debezium Avro events directly from Kafka, joins orders with products, and writes enriched results to a new topic.

In this step, get ready to:

• Create a custom Flink Docker image with Kafka and Avro connectors.
• Add Flink services to Docker Compose and verify the Web UI.
• Write and execute a Flink SQL streaming aggregation job.

Create the Flink Dockerfile

Flink SQL needs connector JARs to read from Kafka and deserialize Avro messages. The base Flink image does not include these, so you will build a custom image that downloads the required JARs at build time.

• Create a new folder called flink inside your project root directory by running:

mkdir flink

• Create a file called Dockerfile inside the flink folder and add the following content:

FROM flink:2.2.0-scala_2.12

# Download Flink SQL Kafka connector
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/4.0.0-2.0/flink-sql-connector-kafka-4.0.0-2.0.jar

# Download Flink Avro Confluent Registry format
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-avro-confluent-registry/2.2.0/flink-avro-confluent-registry-2.2.0.jar

# Download Flink Avro base dependency
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-avro/2.2.0/flink-avro-2.2.0.jar

What does this Dockerfile do?

It extends the official flink:2.2.0-scala_2.12 image and downloads three JARs into Flink's lib directory:

• flink-sql-connector-kafka lets Flink SQL read from and write to Kafka topics.
• flink-avro-confluent-registry enables the debezium-avro-confluent format, which deserializes Debezium CDC events using Schema Registry.
• flink-avro provides the base Avro serialization support that the registry format depends on.

• Save flink/Dockerfile.

Seeing wget failures during build?

Ensure you have internet connectivity. The JARs are downloaded from Maven Central during docker compose build. If your network blocks Maven, download the JARs manually and COPY them in the Dockerfile instead.

help me troubleshoot wget failures in my Flink Dockerfile build

Add Flink services to Docker Compose

Flink runs as a session cluster with three containers: a JobManager that coordinates work, a TaskManager that executes it, and an SQL Client for interactive queries.

• Open docker-compose.yml and add the following three services after the kafka-connect service:

  flink-jobmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-jobmanager
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        state.checkpoints.dir: file:///tmp/flink-checkpoints
        execution.checkpointing.interval: 10s

  flink-taskmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-taskmanager
    container_name: flink-taskmanager
    depends_on:
      - flink-jobmanager
    command: taskmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        taskmanager.numberOfTaskSlots: 4

  flink-sql-client:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-sql-client
    container_name: flink-sql-client
    depends_on:
      - flink-jobmanager
    command: bin/sql-client.sh
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        rest.address: flink-jobmanager

What do these services do?

• flink-jobmanager coordinates job scheduling, checkpointing, and recovery. Checkpointing is set to every 10 seconds for exactly-once processing guarantees.
• flink-taskmanager runs the actual stream processing operators. Four task slots means it can execute up to four parallel operator instances.
• flink-sql-client provides an interactive SQL shell for submitting streaming queries to the cluster.

Port conflict with Schema Registry

Both Schema Registry and Flink JobManager use port 8081 internally. Since Schema Registry already binds host port 8081, you need to change Schema Registry's host port mapping to avoid a conflict.

In docker-compose.yml, find the schema-registry service's ports section and change "8081:8081" to "8082:8081". Other services connect to Schema Registry via Docker networking at schema-registry:8081 (the container port), so this change only affects host access.

• Save docker-compose.yml.
• Rebuild and restart the full stack by running:

docker compose up -d --build

• Wait about 30 seconds for all services to start, then verify the Flink Web UI is accessible by opening http://localhost:8081 in your browser.
• You should see the Flink Dashboard showing 1 available TaskManager and 4 task slots.

Flink Web UI not loading?

• Check if the container started by running docker compose ps and looking for flink-jobmanager with status Up.
• If you see a port binding error, confirm you changed Schema Registry's host port to 8082:8081 as described above.
• Run docker compose logs flink-jobmanager to check for startup errors.

help me troubleshoot why the Flink Web UI is not loading at localhost:8081

Create and run the Flink SQL streaming job

Now you will write SQL that tells Flink to continuously read CDC events from Kafka, join orders with products, and aggregate revenue by category every minute.

• Create a file called flink-sql-job.sql inside the flink folder. This file serves as documentation of your streaming job.
• Add the orders CDC source table definition:

-- Create source table reading CDC events from Kafka orders topic
CREATE TABLE orders_cdc (
    order_id INT,
    customer_id INT,
    product_id INT,
    quantity INT,
    total_amount DECIMAL(10, 2),
    status STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'cdc.public.orders',
    'properties.bootstrap.servers' = 'kafka:29092',
    'properties.group.id' = 'flink-cdc-consumer',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'debezium-avro-confluent',
    'debezium-avro-confluent.url' = 'http://schema-registry:8081'
);

What does this table definition do?

• It creates a virtual table backed by the cdc.public.orders Kafka topic.
• The debezium-avro-confluent format tells Flink to deserialize messages as Debezium CDC events encoded in Avro, using Schema Registry for the schema lookup.
• PRIMARY KEY (order_id) NOT ENFORCED marks this as a changelog stream where each key represents a unique row that can be inserted, updated, or deleted.

• Below the orders table, add the products CDC source table:

-- Create source table for products (used for enrichment joins)
CREATE TABLE products_cdc (
    product_id INT,
    name STRING,
    category STRING,
    price DECIMAL(10, 2),
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'cdc.public.products',
    'properties.bootstrap.servers' = 'kafka:29092',
    'properties.group.id' = 'flink-cdc-consumer',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'debezium-avro-confluent',
    'debezium-avro-confluent.url' = 'http://schema-registry:8081'
);

Why a separate products table?

The streaming aggregation needs to know each product's category. By defining products_cdc as a CDC-backed table, Flink maintains an up-to-date view of products that can be used in temporal joins. If a product's category changes, the join reflects the latest state.

• Add the sink table that writes aggregated results to a new Kafka topic:

-- Create sink table for enriched aggregation results
CREATE TABLE order_revenue_by_category (
    category STRING,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    total_revenue DECIMAL(10, 2),
    order_count BIGINT,
    PRIMARY KEY (category, window_start) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'order_revenue_by_category',
    'properties.bootstrap.servers' = 'kafka:29092',
    'format' = 'json'
);

Why JSON format for the sink?

The Snowflake connector (which you will configure in the next step) reads from this topic using a simple JSON converter. Writing aggregation results as JSON keeps the downstream integration straightforward.

• Finally, add the streaming INSERT query that performs the aggregation:

-- Run streaming aggregation: revenue by product category per 1-minute tumbling window
INSERT INTO order_revenue_by_category
SELECT
    p.category,
    TUMBLE_START(o.updated_at, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(o.updated_at, INTERVAL '1' MINUTE) AS window_end,
    SUM(o.total_amount) AS total_revenue,
    COUNT(*) AS order_count
FROM orders_cdc o
JOIN products_cdc FOR SYSTEM_TIME AS OF o.updated_at AS p
    ON o.product_id = p.product_id
WHERE o.status = 'completed'
GROUP BY p.category, TUMBLE(o.updated_at, INTERVAL '1' MINUTE);

How does this streaming query work?

• TUMBLE(o.updated_at, INTERVAL '1' MINUTE) creates 1-minute time windows based on the order's update timestamp.
• JOIN products_cdc FOR SYSTEM_TIME AS OF o.updated_at is a temporal join. It looks up the product's category as it existed at the time the order was updated.
• Only orders with status = 'completed' are counted toward revenue.
• Results flow continuously into the order_revenue_by_category Kafka topic as JSON.

• Save flink/flink-sql-job.sql.

Now execute these statements in the Flink SQL Client.

• Open the Flink SQL Client by running:

docker compose run flink-sql-client

You should see the Flink SQL welcome banner with a command prompt.

• Copy and paste the CREATE TABLE orders_cdc statement from your SQL file into the client and press Enter.
• Copy and paste the CREATE TABLE products_cdc statement and press Enter.
• Copy and paste the CREATE TABLE order_revenue_by_category statement and press Enter.
• Copy and paste the INSERT INTO order_revenue_by_category query and press Enter.

Flink submits the INSERT as a background streaming job. You should see a message indicating the job has been submitted.

• Verify the job is running by opening http://localhost:8081 in your browser and checking the **Running Jobs** section. You should see one job listed.
• Confirm the output topic was created by running:

docker compose exec kafka kafka-topics --bootstrap-server kafka:29092 --list | grep order_revenue

You should see order_revenue_by_category in the output.

Job failed or no topic created?

• Check that Schema Registry is reachable from inside the Docker network. Run docker compose exec flink-jobmanager curl http://schema-registry:8081/subjects to confirm connectivity.
• Verify the Debezium connector is still running with curl http://localhost:8083/connectors/postgres-cdc-source/status. If it is not RUNNING, the source topics may be empty.
• In the Flink Web UI, click on the failed job to see the exception message. Common issues include mismatched schema field names or missing JAR dependencies.

help me debug why my Flink SQL job failed when reading Debezium Avro events from Kafka

✔️ Awesome, I've got everything!

Great. Double check you have saved both flink/Dockerfile and flink/flink-sql-job.sql, and that your Flink job shows as running in the Web UI.

ⓧ I'd like to double check the full code

Here is the complete flink/Dockerfile:

FROM flink:2.2.0-scala_2.12

# Download Flink SQL Kafka connector
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/4.0.0-2.0/flink-sql-connector-kafka-4.0.0-2.0.jar

# Download Flink Avro Confluent Registry format
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-avro-confluent-registry/2.2.0/flink-avro-confluent-registry-2.2.0.jar

# Download Flink Avro base dependency
RUN wget -P /opt/flink/lib \
    https://repo.maven.apache.org/maven2/org/apache/flink/flink-avro/2.2.0/flink-avro-2.2.0.jar

Here is the complete flink/flink-sql-job.sql:

-- Create source table reading CDC events from Kafka orders topic
CREATE TABLE orders_cdc (
    order_id INT,
    customer_id INT,
    product_id INT,
    quantity INT,
    total_amount DECIMAL(10, 2),
    status STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'cdc.public.orders',
    'properties.bootstrap.servers' = 'kafka:29092',
    'properties.group.id' = 'flink-cdc-consumer',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'debezium-avro-confluent',
    'debezium-avro-confluent.url' = 'http://schema-registry:8081'
);

-- Create source table for products (used for enrichment joins)
CREATE TABLE products_cdc (
    product_id INT,
    name STRING,
    category STRING,
    price DECIMAL(10, 2),
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'cdc.public.products',
    'properties.bootstrap.servers' = 'kafka:29092',
    'properties.group.id' = 'flink-cdc-consumer',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'debezium-avro-confluent',
    'debezium-avro-confluent.url' = 'http://schema-registry:8081'
);

-- Create sink table for enriched aggregation results
CREATE TABLE order_revenue_by_category (
    category STRING,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    total_revenue DECIMAL(10, 2),
    order_count BIGINT,
    PRIMARY KEY (category, window_start) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'order_revenue_by_category',
    'properties.bootstrap.servers' = 'kafka:29092',
    'format' = 'json'
);

-- Run streaming aggregation: revenue by product category per 1-minute tumbling window
INSERT INTO order_revenue_by_category
SELECT
    p.category,
    TUMBLE_START(o.updated_at, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(o.updated_at, INTERVAL '1' MINUTE) AS window_end,
    SUM(o.total_amount) AS total_revenue,
    COUNT(*) AS order_count
FROM orders_cdc o
JOIN products_cdc FOR SYSTEM_TIME AS OF o.updated_at AS p
    ON o.product_id = p.product_id
WHERE o.status = 'completed'
GROUP BY p.category, TUMBLE(o.updated_at, INTERVAL '1' MINUTE);

Here are the Flink services added to docker-compose.yml:

  flink-jobmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-jobmanager
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        state.checkpoints.dir: file:///tmp/flink-checkpoints
        execution.checkpointing.interval: 10s

  flink-taskmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-taskmanager
    container_name: flink-taskmanager
    depends_on:
      - flink-jobmanager
    command: taskmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        taskmanager.numberOfTaskSlots: 4

  flink-sql-client:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-sql-client
    container_name: flink-sql-client
    depends_on:
      - flink-jobmanager
    command: bin/sql-client.sh
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        rest.address: flink-jobmanager

Your Flink streaming job is now continuously aggregating revenue by product category. Next up, you will connect this enriched data to Snowflake for cloud-based analytics.

Connect to Snowflake

Your Flink SQL job is producing enriched revenue aggregations to a Kafka topic. The missing piece is getting that data out of Kafka and into a cloud data warehouse where analysts can actually query it.

In this step, you'll register the Snowflake Kafka Connector as a sink that consumes from the order_revenue_by_category topic and delivers records into Snowflake using Snowpipe Streaming for low-latency, exactly-once ingestion. Then you'll prove the full pipeline works end-to-end by tracing changes from PostgreSQL all the way through to Snowflake.

In this step, get ready to:

• Register the Snowflake sink connector via the Kafka Connect REST API.
• Verify data lands in Snowflake within seconds of being produced.
• Trace INSERT, UPDATE, and DELETE operations end-to-end through the full pipeline.

Register the Snowflake sink connector

The Snowflake Kafka Connector is a sink connector that reads from Kafka topics and writes directly into Snowflake tables. You'll configure it with key-pair authentication (the RSA keys you generated in Step 1) and Snowpipe Streaming mode for low-latency delivery.

First, you need your Snowflake account URL. In the Snowflake web UI, look at the URL in your browser's address bar. It follows the pattern your-account-id.snowflakecomputing.com.

• Create the Snowflake sink connector config file by creating connectors/snowflake-sink.json with the following content:

{
  "name": "snowflake-sink",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "tasks.max": "1",
    "topics": "order_revenue_by_category",
    "snowflake.url.name": "[[SNOWFLAKE_URL=\"your-account-id.snowflakecomputing.com\"]]",
    "snowflake.user.name": "CDC_KAFKA_USER",
    "snowflake.private.key": "${file:/etc/kafka-connect/secrets/snowflake-private-key.txt:private_key}",
    "snowflake.database.name": "CDC_PIPELINE",
    "snowflake.schema.name": "REALTIME",
    "snowflake.role.name": "CDC_KAFKA_ROLE",
    "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "buffer.flush.time": "10"
  }
}

What does this configuration do?

This connector config tells Kafka Connect how to deliver data from Kafka into Snowflake.

• topics points to the enriched aggregation topic that Flink writes to.
• snowflake.private.key uses Kafka Connect's config provider to read the private key from a mounted file, avoiding secrets in the config itself.
• snowflake.ingestion.method set to SNOWPIPE_STREAMING provides exactly-once delivery with sub-10-second latency.
• key.converter and value.converter override the worker defaults because Flink writes plain JSON (not Avro) to this topic.
• buffer.flush.time of 10 seconds means the connector flushes buffered records to Snowflake every 10 seconds at most.

• Save connectors/snowflake-sink.json.
• Register the connector with Kafka Connect by running this command:

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

You should see a JSON response echoing back the connector name and config, confirming it was accepted.

• Confirm the connector is running by checking its status:

curl -s http://localhost:8083/connectors/snowflake-sink/status | jq .

You should see "state": "RUNNING" for both the connector and its task.

Connector showing FAILED?

• Check that your snowflake.url.name matches your actual Snowflake account identifier (visible in your browser URL bar when logged into Snowsight).
• Verify that secrets/snowflake-private-key.txt contains only the key body (no BEGIN/END header lines, no newlines). The fat JAR v3.5.2 expects an unencrypted, single-line private key.
• Make sure the RSA public key assigned to CDC_KAFKA_USER in Snowflake matches the private key file. Run DESC USER CDC_KAFKA_USER in Snowsight to check.
• Check the Kafka Connect logs for more detail:

docker logs kafka-connect --tail 50

Still stuck? Why is my Snowflake sink connector in FAILED state?

Verify data is landing in Snowflake

The Snowflake connector auto-creates the target table when it receives its first record. Your Flink job has been producing aggregation results to the order_revenue_by_category topic since the previous step, so data should begin appearing in Snowflake within seconds of the connector starting.

• Open the Snowflake web UI (Snowsight).
• Click Worksheets in the left sidebar, then click the + Worksheet button to create a new SQL worksheet.
• Query the auto-created table by running this SQL:

USE DATABASE CDC_PIPELINE;
USE SCHEMA REALTIME;
USE WAREHOUSE CDC_WAREHOUSE;

SELECT * FROM ORDER_REVENUE_BY_CATEGORY
ORDER BY RECORD_METADATA:CreateTime::TIMESTAMP DESC
LIMIT 10;

You should see rows containing your revenue aggregation data. The table has a RECORD_CONTENT column (the JSON payload from Kafka) and a RECORD_METADATA column (Kafka metadata including offset and timestamp).

Why does the table look different from my Flink output?

The Snowflake connector stores each Kafka record as a semi-structured JSON object in a VARIANT column. This preserves the original data without requiring a predefined schema in Snowflake.

You can extract individual fields using Snowflake's JSON path notation. For example: RECORD_CONTENT:category::STRING extracts the category field as a string.

• Run a more targeted query to see the enriched aggregation fields:

SELECT
  RECORD_CONTENT:category::STRING AS category,
  RECORD_CONTENT:total_revenue::DECIMAL(10,2) AS total_revenue,
  RECORD_CONTENT:order_count::INT AS order_count,
  RECORD_CONTENT:window_start::TIMESTAMP AS window_start
FROM ORDER_REVENUE_BY_CATEGORY
ORDER BY window_start DESC
LIMIT 10;

You should see rows with category names (Electronics, Sports, Home), revenue totals, and order counts grouped by 1-minute windows.

Table not found or no rows?

• The table auto-creates only after the first record arrives. Wait 10-15 seconds after the connector shows RUNNING, then retry the query.
• Confirm the Flink job is still running and producing records by checking the Kafka topic:

docker exec kafka kafka-console-consumer --bootstrap-server kafka:29092 --topic order_revenue_by_category --from-beginning --max-messages 3

If you see JSON messages in the topic but nothing in Snowflake, the connector may have an authentication issue. Check docker logs kafka-connect --tail 100 for errors mentioning the private key or RSA.

Need help? Help me debug why data isn't appearing in my Snowflake table.

Trace changes end-to-end through the pipeline

The real test of a CDC pipeline is proving that changes flow from the source database all the way to the analytics warehouse. You'll perform INSERT, UPDATE, and DELETE operations in PostgreSQL and watch them propagate through Debezium, Kafka, Flink, and into Snowflake.

• Connect to PostgreSQL by running this command:

docker exec -it postgres psql -U cdc_user -d ecommerce

• Insert a new completed order by running this SQL:

-- Insert a new completed order (triggers CDC -> Flink aggregation -> Snowflake)
INSERT INTO orders (customer_id, product_id, quantity, total_amount, status)
VALUES (2, 1, 3, 239.97, 'completed');

What happens when you insert this row?

• PostgreSQL writes the change to its Write-Ahead Log (WAL).
• Debezium reads the WAL and produces an Avro CDC event to the cdc.public.orders topic.
• Flink SQL joins the order with the products table, sees status = 'completed', and includes it in the revenue aggregation.
• Flink writes the updated aggregation result (JSON) to the order_revenue_by_category topic.
• The Snowflake connector picks up the new record and delivers it via Snowpipe Streaming.

• Update an existing order to mark it as completed:

-- Update a pending order to completed (triggers a new aggregation window result)
UPDATE orders SET status = 'completed', updated_at = CURRENT_TIMESTAMP
WHERE order_id = 3;

• Delete a cancelled order:

-- Delete an order (Debezium emits a delete event)
DELETE FROM orders WHERE order_id = 4;

• Exit the PostgreSQL prompt:

\q

• Wait about 15-20 seconds for the changes to propagate through the pipeline.
• Go back to your Snowflake worksheet and re-run the query to check for new rows:

SELECT
  RECORD_CONTENT:category::STRING AS category,
  RECORD_CONTENT:total_revenue::DECIMAL(10,2) AS total_revenue,
  RECORD_CONTENT:order_count::INT AS order_count,
  RECORD_CONTENT:window_start::TIMESTAMP AS window_start
FROM ORDER_REVENUE_BY_CATEGORY
ORDER BY window_start DESC
LIMIT 10;

You should see new rows reflecting the updated aggregation totals. The Electronics category total should now include the $239.97 order you inserted, and the Sports category should reflect the order you updated to completed status.

Why doesn't the DELETE show up as a separate row?

Your Flink SQL aggregation only counts orders with status = 'completed'. A DELETE on a pending or cancelled order doesn't change the completed revenue totals. The DELETE event still flows through Debezium and Kafka, but Flink's WHERE clause filters it out before aggregation.

Not seeing updated data in Snowflake?

• The pipeline has multiple stages with buffering at each. Allow up to 60 seconds for end-to-end propagation (Flink's 1-minute tumbling window means results emit at window boundaries).
• Verify the Debezium connector is still RUNNING:

curl -s http://localhost:8083/connectors/postgres-cdc-source/status | jq .connector.state

• Verify the Snowflake connector is still RUNNING:

curl -s http://localhost:8083/connectors/snowflake-sink/status | jq .connector.state

Both should return "RUNNING". If either is FAILED, check docker logs kafka-connect for the specific error.

Need help tracing the issue? Help me debug my end-to-end CDC pipeline.

✔️ Awesome, I've got everything!

Great. Double check you've saved your connector config file and that both connectors show RUNNING status.

ⓧ I'd like to double check the full code

Here is the complete Snowflake sink connector configuration:

{
  "name": "snowflake-sink",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "tasks.max": "1",
    "topics": "order_revenue_by_category",
    "snowflake.url.name": "your-account-id.snowflakecomputing.com",
    "snowflake.user.name": "CDC_KAFKA_USER",
    "snowflake.private.key": "${file:/etc/kafka-connect/secrets/snowflake-private-key.txt:private_key}",
    "snowflake.database.name": "CDC_PIPELINE",
    "snowflake.schema.name": "REALTIME",
    "snowflake.role.name": "CDC_KAFKA_ROLE",
    "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "buffer.flush.time": "10"
  }
}

Your full CDC pipeline is now streaming data from PostgreSQL through Kafka and Flink into Snowflake. Next up, you'll add observability with Prometheus and Grafana so you can monitor the health of every component in real-time.

Add Observability with Prometheus and Grafana

Your pipeline is now streaming data end-to-end from PostgreSQL through Kafka and Flink into Snowflake. But how do you know if something breaks at 2 AM?

Production pipelines need observability. In this step, you will add Prometheus for metrics collection and Grafana for visualization. Together, they give you a real-time dashboard showing consumer lag, connector health, and processing throughput.

In this step, get ready to:

• Add Prometheus and Grafana services to Docker Compose.
• Create a Prometheus configuration to scrape Kafka, Kafka Connect, and Flink metrics.
• Build a Grafana dashboard with panels for pipeline health monitoring.

Add Prometheus and Grafana to Docker Compose

Your Kafka container already exposes JMX metrics on port 9101 (configured with KAFKA_JMX_PORT in Step 2). Prometheus will scrape these metrics endpoints directly.

• Open docker-compose.yml in your editor.
• Add the following two services at the bottom of the file, after the flink-sql-client service:

  prometheus:
    image: prom/prometheus:v2.53.0
    hostname: prometheus
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:11.1.0
    hostname: grafana
    container_name: grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin
    volumes:
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning

What do these services do?

• Prometheus (prom/prometheus:v2.53.0) scrapes metrics from your other containers on a schedule and stores time-series data. It exposes a query UI at port 9090.
• Grafana (grafana/grafana:11.1.0) connects to Prometheus as a data source and renders dashboards. The GF_SECURITY_ADMIN_PASSWORD environment variable sets the admin login password.
• Both services mount local configuration directories so you can define scrape targets and dashboards as code.

• Save docker-compose.yml.

YAML indentation error?

Make sure the prometheus and grafana service definitions are indented at the same level as your other services (two spaces under services:). YAML is sensitive to whitespace.

help me fix YAML indentation in my docker-compose.yml

Create the Prometheus scrape configuration

Prometheus needs a configuration file that tells it which endpoints to scrape and how often. You will create this file in the monitoring directory that the Docker volume mount expects.

• Create the monitoring directory structure by running these commands:

mkdir -p monitoring/grafana/provisioning/datasources
mkdir -p monitoring/grafana/provisioning/dashboards

• Create the Prometheus configuration file by running:

touch monitoring/prometheus.yml

• Open monitoring/prometheus.yml and add the following configuration:

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "kafka"
    static_configs:
      - targets: ["kafka:9101"]

  - job_name: "kafka-connect"
    metrics_path: /metrics
    static_configs:
      - targets: ["kafka-connect:8083"]

  - job_name: "flink"
    static_configs:
      - targets: ["flink-jobmanager:9249"]

What does this configuration do?

• scrape_interval: 15s tells Prometheus to pull metrics from all targets every 15 seconds.
• The kafka job scrapes JMX metrics from the broker on port 9101. This includes message rates, request latency, and broker health.
• The kafka-connect job scrapes the Connect REST API's metrics endpoint. This exposes connector task status and consumer lag for Connect workers.
• The flink job scrapes the Flink JobManager's metrics reporter on port 9249. This shows job throughput and checkpoint status.

Now create the Grafana datasource provisioning file so Grafana automatically connects to Prometheus on startup.

• Create the datasource configuration by adding this content to monitoring/grafana/provisioning/datasources/datasource.yml:

apiVersion: 1

datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true
    editable: false

Why provision the datasource as code?

Grafana reads YAML files from its provisioning directory on startup and automatically configures data sources. This means anyone who clones your repo and runs docker compose up gets a fully configured Grafana without manual setup.

The url: http://prometheus:9090 uses the Docker network hostname. Grafana and Prometheus are on the same Docker network, so they can reach each other by container name.

• Save all files.
• Restart the Docker Compose stack to bring up the new services by running:

docker compose up -d

• Wait about 30 seconds for the new containers to start, then verify Prometheus is running by opening http://localhost:9090 in your browser.
• Click Status in the top navigation bar, then click Targets.
• You should see three targets listed: kafka, kafka-connect, and flink. Their state should show as UP (green).

Targets showing as DOWN?

If a target shows as DOWN, the container it points to may not be fully started yet. Wait another 30 seconds and refresh the Targets page. If the flink target stays DOWN, this is expected if Flink does not expose a Prometheus metrics reporter by default. The kafka and kafka-connect targets should be green.

Why are my Prometheus scrape targets showing as DOWN?

Create a Grafana pipeline health dashboard

With Prometheus collecting metrics, you can now build a Grafana dashboard that visualizes your pipeline health at a glance.

• Open Grafana at http://localhost:3000 in your browser.
• Log in with username admin and password admin.
• Grafana will prompt you to change the password. Click Skip for now.

Why is Prometheus already listed as a datasource?

The provisioning file you created earlier (datasource.yml) configured this automatically on startup. Grafana loaded it from the mounted volume at /etc/grafana/provisioning.

Now create a dashboard with four panels that monitor the critical signals of your CDC pipeline.

• Click the + icon in the left sidebar, then click New dashboard.
• Click Add visualization.
• Select Prometheus as the data source.

Create the first panel for Kafka consumer lag.

• In the Metric field, enter this PromQL query:

kafka_consumer_fetch_manager_metrics_records_lag_max

• Set the panel title to Kafka Consumer Lag in the right sidebar under Title.
• Click Apply in the top right to save the panel.

Add a second panel for Kafka Connect connector status.

• Click Add in the top toolbar, then Visualization.
• Select Prometheus as the data source.
• Enter this query for connector task status:

kafka_connect_connector_status

• Set the panel title to Kafka Connect Connector Status.
• Click Apply.

Add a third panel for Flink job throughput.

• Add another visualization panel with Prometheus as the data source.
• Enter this query:

flink_taskmanager_job_task_numRecordsOutPerSecond

• Set the panel title to Flink Job Throughput (records/sec).
• Click Apply.

Add a fourth panel for end-to-end pipeline latency.

• Add one more visualization panel with Prometheus as the data source.
• Enter this query for message production latency:

kafka_server_BrokerTopicMetrics_TotalProduceRequestsPerSec

• Set the panel title to End-to-End Pipeline Latency.
• Click Apply.

Save the complete dashboard.

• Click the Save dashboard icon (floppy disk) in the top toolbar.
• Enter CDC Pipeline Health as the dashboard name.
• Click Save.

What do these panels tell you?

• Kafka Consumer Lag shows how far behind consumers are from producers. A growing lag means your pipeline cannot keep up with incoming data.
• Kafka Connect Connector Status reports whether your Debezium source and Snowflake sink connectors are running or failed.
• Flink Job Throughput shows how many records per second Flink processes. A drop to zero means your streaming aggregation stopped.
• End-to-End Pipeline Latency tracks the produce request rate at the broker level, giving you a proxy for how quickly data moves through the system.

Some panels may show 'No data'

Not all metrics are available immediately. The Flink metrics endpoint (port 9249) requires the Prometheus metrics reporter to be explicitly enabled in Flink's configuration. The JMX-based Kafka metrics will populate within 15 seconds of Prometheus starting. If panels show no data, try running the data generator (next step) to produce activity.

• Verify your dashboard displays at least one panel with data by checking the Kafka Consumer Lag or Kafka Connect Connector Status panels.

Grafana shows 'No data' on all panels?

First check that Prometheus is scraping successfully by visiting http://localhost:9090/targets. If the targets are UP but Grafana shows nothing, click the Explore icon in Grafana's sidebar and try typing up as a query. You should see results with value 1 for each target.

Why is my Grafana dashboard showing no data from Prometheus?

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml, monitoring/prometheus.yml, and monitoring/grafana/provisioning/datasources/datasource.yml.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml including the new Prometheus and Grafana services:

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.8
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092"
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:29093"
      KAFKA_LISTENERS: "PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092"
      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LOG_DIRS: "/tmp/kraft-combined-logs"
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
    healthcheck:
      test: kafka-topics --bootstrap-server kafka:29092 --list
      interval: 10s
      timeout: 5s
      retries: 5

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.8
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "kafka:29092"
      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"

  postgres:
    image: postgres:16
    hostname: postgres
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: cdc_user
      POSTGRES_PASSWORD: cdc_password
      POSTGRES_DB: ecommerce
    command: >
      postgres
      -c wal_level=logical
      -c max_wal_senders=4
      -c max_replication_slots=4
    volumes:
      - ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql

  kafka-connect:
    build:
      context: ./connect
      dockerfile: Dockerfile
    hostname: kafka-connect
    container_name: kafka-connect
    depends_on:
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_started
      postgres:
        condition: service_started
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka:29092"
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
      CONNECT_GROUP_ID: cdc-connect-cluster
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
    volumes:
      - ./secrets:/etc/kafka-connect/secrets

  flink-jobmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-jobmanager
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        state.checkpoints.dir: file:///tmp/flink-checkpoints
        execution.checkpointing.interval: 10s

  flink-taskmanager:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-taskmanager
    container_name: flink-taskmanager
    depends_on:
      - flink-jobmanager
    command: taskmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        taskmanager.numberOfTaskSlots: 4

  flink-sql-client:
    build:
      context: ./flink
      dockerfile: Dockerfile
    hostname: flink-sql-client
    container_name: flink-sql-client
    depends_on:
      - flink-jobmanager
    command: bin/sql-client.sh
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        rest.address: flink-jobmanager

  prometheus:
    image: prom/prometheus:v2.53.0
    hostname: prometheus
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:11.1.0
    hostname: grafana
    container_name: grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin
    volumes:
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning

Here is the complete monitoring/prometheus.yml:

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "kafka"
    static_configs:
      - targets: ["kafka:9101"]

  - job_name: "kafka-connect"
    metrics_path: /metrics
    static_configs:
      - targets: ["kafka-connect:8083"]

  - job_name: "flink"
    static_configs:
      - targets: ["flink-jobmanager:9249"]

Here is the Grafana datasource provisioning file monitoring/grafana/provisioning/datasources/datasource.yml:

apiVersion: 1

datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true
    editable: false

You now have full observability over your CDC pipeline. Next up, you will put the pipeline under load and prove it handles schema evolution gracefully.

Load Test and Schema Evolution

Your observability stack is live and ready to show you what happens inside your pipeline. Now it is time to put the system under real pressure.

In this step, you will build a Python script that simulates continuous e-commerce traffic, generating INSERTs, UPDATEs, and DELETEs at a configurable rate. You will watch the Grafana dashboard respond in real-time. Then you will prove the pipeline handles schema evolution gracefully by adding a new column to PostgreSQL and watching it propagate all the way to Snowflake without any downtime.

In this step, get ready to:

• Create a Python data generator that simulates realistic e-commerce transactions at a configurable rate.
• Load-test the pipeline and observe Grafana metrics updating in real-time.
• Demonstrate schema evolution by adding a column to PostgreSQL and verifying it propagates without breaking the pipeline.

Create the data generator script

A streaming pipeline is only as good as the data flowing through it. You need a script that continuously generates realistic database operations so you can observe how the pipeline handles sustained load.

• Install the Python PostgreSQL driver by running:

pip install psycopg2-binary

Install failing?

If you see a build error on Linux, install the system library first with sudo apt-get install libpq-dev. On macOS with Homebrew, run brew install postgresql. Help me install psycopg2 on my operating system.

• Create the file scripts/generate_data.py and add the configuration section by pasting this code:

import psycopg2
import random
import time
import sys

# Configuration
DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "cdc_user",
    "password": "cdc_password"
}

CATEGORIES = ["Electronics", "Sports", "Home"]
STATUSES = ["pending", "completed", "cancelled"]

def get_connection():
    return psycopg2.connect(**DB_CONFIG)

What does this code do?

DB_CONFIG connects to the same PostgreSQL instance your Docker Compose stack exposes on port 5432. The script runs on your host machine, not inside Docker, so it uses localhost.

get_connection() is a helper that creates a fresh database connection using the config dictionary.

• Below the get_connection() function, add the create_order() function that inserts new orders:

def create_order(conn):
    cur = conn.cursor()
    customer_id = random.randint(1, 3)
    product_id = random.randint(1, 5)
    quantity = random.randint(1, 5)

    cur.execute("SELECT price FROM products WHERE product_id = %s", (product_id,))
    price = cur.fetchone()[0]
    total_amount = float(price) * quantity

    cur.execute(
        """INSERT INTO orders (customer_id, product_id, quantity, total_amount, status)
           VALUES (%s, %s, %s, %s, 'pending') RETURNING order_id""",
        (customer_id, product_id, quantity, total_amount)
    )
    order_id = cur.fetchone()[0]
    conn.commit()
    print(f"[CREATE] Order #{order_id}: customer={customer_id}, product={product_id}, total=${total_amount:.2f}")
    return order_id

What does this code do?

This function picks a random customer and product, looks up the product price, calculates the total, and inserts a new order with status pending. Each INSERT triggers a CDC event that Debezium captures from the WAL.

• Add the update_order_status() function below create_order():

def update_order_status(conn):
    cur = conn.cursor()
    cur.execute("SELECT order_id FROM orders WHERE status = 'pending' ORDER BY RANDOM() LIMIT 1")
    row = cur.fetchone()
    if row:
        order_id = row[0]
        new_status = random.choice(["completed", "cancelled"])
        cur.execute(
            "UPDATE orders SET status = %s, updated_at = CURRENT_TIMESTAMP WHERE order_id = %s",
            (new_status, order_id)
        )
        conn.commit()
        print(f"[UPDATE] Order #{order_id} -> {new_status}")

What does this code do?

This picks a random pending order and moves it to either completed or cancelled. UPDATEs generate CDC events that include both the before and after state, which is why you set REPLICA IDENTITY FULL earlier.

• Add the delete_order() function below update_order_status():

def delete_order(conn):
    cur = conn.cursor()
    cur.execute("SELECT order_id FROM orders WHERE status = 'cancelled' ORDER BY RANDOM() LIMIT 1")
    row = cur.fetchone()
    if row:
        order_id = row[0]
        cur.execute("DELETE FROM orders WHERE order_id = %s", (order_id,))
        conn.commit()
        print(f"[DELETE] Order #{order_id} removed")

What does this code do?

This removes a random cancelled order. DELETEs produce tombstone events in Kafka, which downstream consumers use to know a record was removed. Debezium publishes these because you configured tombstones.on.delete: true in the source connector.

• Finally, add the main() function and entry point at the bottom of the file:

def main():
    rate = float(sys.argv[1]) if len(sys.argv) > 1 else 1.0
    print(f"Starting data generator at {rate} events/second...")
    print("Press Ctrl+C to stop.\n")

    conn = get_connection()
    try:
        while True:
            action = random.choices(
                ["create", "update", "delete"],
                weights=[60, 30, 10],
                k=1
            )[0]

            if action == "create":
                create_order(conn)
            elif action == "update":
                update_order_status(conn)
            else:
                delete_order(conn)

            time.sleep(1.0 / rate)
    except KeyboardInterrupt:
        print("\nStopping data generator.")
    finally:
        conn.close()

if __name__ == "__main__":
    main()

What does this code do?

The main() function reads an optional rate argument from the command line (defaulting to 1 event per second). It loops continuously, picking a random action with weighted probability: 60% creates, 30% updates, 10% deletes.

This distribution mirrors real e-commerce traffic where most activity is new orders, some orders get fulfilled, and a small percentage get cancelled and removed.

• Save scripts/generate_data.py.

Not sure about the file location?

The file should be at scripts/generate_data.py inside your project root (the same folder that contains docker-compose.yml). If you do not have a scripts/ folder yet, create it first. Help me figure out where to put the generate_data.py file in my project structure.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

import psycopg2
import random
import time
import sys

# Configuration
DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "cdc_user",
    "password": "cdc_password"
}

CATEGORIES = ["Electronics", "Sports", "Home"]
STATUSES = ["pending", "completed", "cancelled"]

def get_connection():
    return psycopg2.connect(**DB_CONFIG)

def create_order(conn):
    cur = conn.cursor()
    customer_id = random.randint(1, 3)
    product_id = random.randint(1, 5)
    quantity = random.randint(1, 5)

    cur.execute("SELECT price FROM products WHERE product_id = %s", (product_id,))
    price = cur.fetchone()[0]
    total_amount = float(price) * quantity

    cur.execute(
        """INSERT INTO orders (customer_id, product_id, quantity, total_amount, status)
           VALUES (%s, %s, %s, %s, 'pending') RETURNING order_id""",
        (customer_id, product_id, quantity, total_amount)
    )
    order_id = cur.fetchone()[0]
    conn.commit()
    print(f"[CREATE] Order #{order_id}: customer={customer_id}, product={product_id}, total=${total_amount:.2f}")
    return order_id

def update_order_status(conn):
    cur = conn.cursor()
    cur.execute("SELECT order_id FROM orders WHERE status = 'pending' ORDER BY RANDOM() LIMIT 1")
    row = cur.fetchone()
    if row:
        order_id = row[0]
        new_status = random.choice(["completed", "cancelled"])
        cur.execute(
            "UPDATE orders SET status = %s, updated_at = CURRENT_TIMESTAMP WHERE order_id = %s",
            (new_status, order_id)
        )
        conn.commit()
        print(f"[UPDATE] Order #{order_id} -> {new_status}")

def delete_order(conn):
    cur = conn.cursor()
    cur.execute("SELECT order_id FROM orders WHERE status = 'cancelled' ORDER BY RANDOM() LIMIT 1")
    row = cur.fetchone()
    if row:
        order_id = row[0]
        cur.execute("DELETE FROM orders WHERE order_id = %s", (order_id,))
        conn.commit()
        print(f"[DELETE] Order #{order_id} removed")

def main():
    rate = float(sys.argv[1]) if len(sys.argv) > 1 else 1.0
    print(f"Starting data generator at {rate} events/second...")
    print("Press Ctrl+C to stop.\n")

    conn = get_connection()
    try:
        while True:
            action = random.choices(
                ["create", "update", "delete"],
                weights=[60, 30, 10],
                k=1
            )[0]

            if action == "create":
                create_order(conn)
            elif action == "update":
                update_order_status(conn)
            else:
                delete_order(conn)

            time.sleep(1.0 / rate)
    except KeyboardInterrupt:
        print("\nStopping data generator.")
    finally:
        conn.close()

if __name__ == "__main__":
    main()

Run the load test and monitor Grafana

With the script ready, you can now push sustained traffic through the pipeline. Running the generator at 2 events per second gives you enough throughput to see the Grafana metrics respond without overwhelming your local Docker setup.

• Start the data generator at 2 events per second by running:

python scripts/generate_data.py 2

You should see a stream of output like this:

Starting data generator at 2.0 events/second...
Press Ctrl+C to stop.

[CREATE] Order #4: customer=2, product=3, total=$149.97
[CREATE] Order #5: customer=1, product=5, total=$59.98
[UPDATE] Order #4 -> completed
[CREATE] Order #6: customer=3, product=1, total=$239.97
[UPDATE] Order #5 -> cancelled
[DELETE] Order #5 removed

What should I see?

The terminal shows a continuous stream of CREATE, UPDATE, and DELETE operations. Most lines are CREATEs (60% weight), some are UPDATEs (30%), and a few are DELETEs (10%). Each one triggers a CDC event that flows through the entire pipeline.

• Open your browser and navigate to http://localhost:3000 to view your Grafana dashboard.
• Log in with username admin and password admin if prompted.
• Open your pipeline health dashboard.
• Watch the panels for 30-60 seconds while the generator runs.

You should see the consumer lag panel fluctuate as Kafka receives events faster than consumers process them, then settle back down. The throughput panel shows a steady rate of approximately 2 records per second.

• Let the generator run for at least 60 seconds so you accumulate enough data for the 1-minute tumbling windows in your Flink SQL job.
• Verify data is reaching Snowflake by running this query in the Snowflake web UI:

SELECT * FROM CDC_PIPELINE.REALTIME.ORDER_REVENUE_BY_CATEGORY
ORDER BY RECORD_METADATA:CreateTime DESC
LIMIT 10;

You should see new rows appearing with recent timestamps, confirming the full pipeline is processing load test data end-to-end.

• Press Ctrl+C in your terminal to stop the data generator.

Not seeing metrics in Grafana?

Make sure your Prometheus datasource is configured correctly in Grafana. Check that Prometheus is scraping targets by visiting http://localhost:9090/targets in your browser. All targets should show state UP.

If the generator exits immediately with a connection error, confirm your Docker containers are still running with docker compose ps. Help me debug why my data generator cannot connect to PostgreSQL.

Demonstrate schema evolution

In production, database schemas change constantly. New columns get added, types get modified, and fields get deprecated. A production-grade pipeline must handle these changes without downtime or data loss.

Because you configured Avro serialization with Schema Registry, your pipeline can accept backward-compatible schema changes automatically. Adding a new column with a default value is a backward-compatible change.

• Connect to PostgreSQL by running:

docker exec -it postgres psql -U cdc_user -d ecommerce

• Add a new discount column to the orders table by running this SQL:

ALTER TABLE orders ADD COLUMN discount DECIMAL(5,2) DEFAULT 0.00;

Why is this a safe schema change?

Adding a column with a default value is a backward-compatible change in Avro. Existing consumers that do not know about the new field simply ignore it. Schema Registry enforces backward compatibility by default, so it will accept this new schema version without any manual intervention.

• Insert a new order that uses the discount column:

INSERT INTO orders (customer_id, product_id, quantity, total_amount, status, discount)
VALUES (1, 1, 2, 143.98, 'completed', 15.99);

• Exit the PostgreSQL shell by typing \q and pressing Enter.

• Verify Schema Registry accepted the evolved schema by running:

curl -s http://localhost:8081/subjects/cdc.public.orders-value/versions | python3 -m json.tool

You should see a list with at least two version numbers (e.g., [1, 2]). The new version was registered automatically when Debezium captured the first event with the new schema.

• Inspect the latest schema version to confirm the discount field is present:

curl -s http://localhost:8081/subjects/cdc.public.orders-value/versions/latest | python3 -m json.tool

In the schema JSON, you should see a field named discount with a default value. This confirms Schema Registry accepted the compatible change.

• Verify the pipeline did not break by checking connector status:

curl -s http://localhost:8083/connectors/postgres-cdc-source/status | python3 -m json.tool
curl -s http://localhost:8083/connectors/snowflake-sink/status | python3 -m json.tool

Both connectors should show RUNNING state. The schema change did not crash or pause either connector.

• Confirm the data reached Snowflake by running this query in the Snowflake web UI:

SELECT * FROM CDC_PIPELINE.REALTIME.ORDER_REVENUE_BY_CATEGORY
ORDER BY RECORD_METADATA:CreateTime DESC
LIMIT 5;

You should see recent rows that include the revenue from your new order. The pipeline processed the schema change seamlessly.

Why does this work without reconfiguring anything?

Schema Registry enforces backward compatibility by default. The new schema (with the discount column) is backward-compatible because the field has a default value. Consumers using the old schema can still read new messages by simply ignoring the unknown field.

Flink also handles this gracefully. Since the Flink SQL table definition does not include discount, it ignores the extra field and continues aggregating revenue as before. No restart required.

Schema Registry rejected the change?

If Schema Registry returns a compatibility error, the change may not be backward-compatible. Make sure you used DEFAULT 0.00 in the ALTER TABLE statement. Without a default value, Avro treats the field as required, which breaks backward compatibility.

If connectors show FAILED status, check the connector logs with docker logs kafka-connect --tail 50. Help me understand why my schema evolution broke the pipeline.

Secret mission

Dead Letter Queue + Exactly-Once Delivery

Your pipeline works, but what happens when bad data sneaks in? In production, a single malformed record can crash a connector and halt the entire pipeline. This secret mission adds Dead Letter Queue error handling and exactly-once delivery guarantees to make your pipeline truly production-grade.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. The Docker containers run locally at no cost, but your Snowflake trial has a 30-day limit with $400 in credits.

Cost warning

Your Snowflake X-Small warehouse auto-suspends after 5 minutes of inactivity, but any queries or active connectors will consume credits. If you are not actively testing, suspend the warehouse manually:

ALTER WAREHOUSE CDC_WAREHOUSE SUSPEND;

Once your 30-day trial or $400 in credits runs out, your Snowflake account will be frozen.

Resources you used:

• 9 Docker containers (Kafka, Schema Registry, PostgreSQL, Kafka Connect, Flink JobManager, Flink TaskManager, Flink SQL Client, Prometheus, Grafana).
• Docker volumes storing PostgreSQL data and Kafka logs.
• Snowflake trial account with CDC_PIPELINE database, REALTIME schema, CDC_WAREHOUSE, CDC_KAFKA_ROLE, and CDC_KAFKA_USER.
• RSA key pair files in your local secrets folder.
• Python data generator script and project directory on your Desktop.

Keep everything running

No action needed. Choose this if you are still actively experimenting with the pipeline or plan to continue the Secret Mission.

• Your Docker Compose stack stays up and running at no cost.
• Your Snowflake trial continues until credits or 30 days expire.
• The X-Small warehouse auto-suspends after 5 minutes of inactivity, so idle time does not consume credits.

Pause - I'll come back to this later

Shut down running containers to free up memory and CPU without deleting your data. You can resume exactly where you left off.

• Stop all containers by running:

docker compose stop

• When you are ready to resume, run:

docker compose start

• Your Kafka topics, PostgreSQL data, and connector configs are preserved in Docker volumes.
• In Snowflake, suspend the warehouse to avoid burning credits while paused:

ALTER WAREHOUSE CDC_WAREHOUSE SUSPEND;

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes all local data and Snowflake objects.

• Remove all containers, networks, and volumes by running:

docker compose down -v

• In Snowflake, drop the database and warehouse:

DROP DATABASE CDC_PIPELINE;
DROP WAREHOUSE CDC_WAREHOUSE;

• Optionally drop the dedicated role and user:

USE ROLE SECURITYADMIN;
DROP USER IF EXISTS CDC_KAFKA_USER;
DROP ROLE IF EXISTS CDC_KAFKA_ROLE;

• Delete the project folder from your Desktop if you no longer need the source files.
• Delete the RSA key pair files from your secrets folder.

Nice Work!

Nice Work!

You just built a production-grade real-time CDC pipeline that streams database changes from PostgreSQL through Kafka to Snowflake, with schema governance, stream processing, and full observability.

You've learned how to:

• Capture real-time database changes using Debezium CDC and stream them as Avro-serialized events governed by Schema Registry.
• Build real-time streaming aggregations with Flink SQL and sink enriched data to Snowflake using key-pair authentication.
• Monitor pipeline health with Prometheus and Grafana dashboards and prove the pipeline handles schema evolution without breaking.
• Secret Mission: Added Dead Letter Queue error handling and exactly-once delivery guarantees across the entire pipeline.

Ready to quiz yourself?