Real-World Case Study: E-Commerce CDC Pipeline

1. Source: PostgreSQL (RDS)

The production database remained unchanged, except for enabling logical replication:

-- Enable logical replication (requires restart)
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;

-- Create replication slot for Debezium
SELECT pg_create_logical_replication_slot('debezium_slot', 'pgoutput');

-- Grant necessary permissions
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;
GRANT USAGE ON SCHEMA public TO debezium_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium_user;

Key Decision: They used RDS PostgreSQL (not Aurora) because Aurora's logical replication had known limitations with high-throughput workloads at the time.

2. Capture: Debezium PostgreSQL Connector

The Debezium connector reads the PostgreSQL WAL and publishes change events to Kafka:

{
  "name": "orders-postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "shopstream-prod.abcdef.us-east-1.rds.amazonaws.com",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "${file:/secrets/db-password.txt:password}",
    "database.dbname": "orders",
    "topic.prefix": "orders",
    "table.include.list": "public.orders,public.customers,public.line_items,public.products",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_publication",
    "plugin.name": "pgoutput",
    "snapshot.mode": "initial",
    "snapshot.locking.mode": "none",
    "heartbeat.interval.ms": "30000",
    "decimal.handling.mode": "precise",
    "time.precision.mode": "adaptive",
    "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",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

Key Decisions:

• snapshot.locking.mode: none prevents holding locks during initial snapshot
• plugin.name: pgoutput uses PostgreSQL's native logical decoding (no external plugin needed)
• tombstones.on.delete: true ensures deletes are properly propagated
• Avro with Schema Registry for type safety and schema evolution

3. Stream: Kafka (MSK)

They chose AWS Managed Streaming for Kafka (MSK) to reduce operational burden:

• Cluster Size: 3 brokers (m5.large) across 3 AZs
• Storage: 1TB EBS per broker (gp3)
• Topic Config:
  • 3 partitions per topic (orders volume: ~2K events/sec peak)
  • Replication factor: 3 (for durability)
  • Retention: 7 days (compliance requirement)
  • Compression: lz4 (balance of speed and ratio)

Topic Naming Convention: {database}.{schema}.{table}
Example: orders.public.orders, orders.public.customers

4. Sink: Snowflake Connector

The Snowflake Sink connector consumes from Kafka and applies changes to Snowflake:

{
  "name": "snowflake-sink",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "topics": "orders.public.orders,orders.public.customers,orders.public.line_items,orders.public.products",
    "snowflake.url.name": "shopstream.snowflakecomputing.com",
    "snowflake.user.name": "kafka_loader",
    "snowflake.private.key": "${file:/secrets/snowflake-key.pem:key}",
    "snowflake.database.name": "ANALYTICS",
    "snowflake.schema.name": "ORDERS_CDC",
    "snowflake.role.name": "KAFKA_LOADER_ROLE",
    "buffer.count.records": "10000",
    "buffer.size.bytes": "5000000",
    "buffer.flush.time": "60",
    "snowflake.metadata.createtime": "false",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "snowflake.enable.schematization": "true",
    "snowflake.topic2table.map": "orders.public.orders:ORDERS,orders.public.customers:CUSTOMERS"
  }
}

How it works: The connector stages micro-batches to S3, then loads them into Snowflake using COPY INTO. Snowflake then uses a stored procedure to MERGE staged records into the target tables.

5. Transformation: Kafka Streams (Optional)

For real-time aggregations, they built a Kafka Streams application:

// Real-time revenue by category (5-minute tumbling window)
KStream<String, Order> orders = builder.stream("orders.public.orders");
KStream<String, LineItem> lineItems = builder.stream("orders.public.line_items");

// Join orders with line items, aggregate by product category
KTable<Windowed<String>, RevenueMetric> revenueByCategory = orders
    .join(lineItems, 
        (order, lineItem) -> new OrderLineItem(order, lineItem),
        JoinWindows.of(Duration.ofMinutes(5)))
    .groupBy((key, orderLineItem) -> orderLineItem.getCategory())
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .aggregate(
        RevenueMetric::new,
        (key, orderLineItem, agg) -> agg.add(orderLineItem.getTotal()),
        Materialized.with(Serdes.String(), new RevenueMetricSerde())
    );

// Expose via interactive queries for dashboard API
revenueByCategory.toStream()
    .to("revenue-by-category-windowed", Produced.with(WindowedSerdes.timeWindowedSerdeFrom(String.class), new RevenueMetricSerde()));