Change Data Capture (CDC) — Deep Dive

MINIMAL:
  Records only what is needed for crash recovery.
  Does NOT record enough information for logical decoding. Cannot be used for CDC.
  This is the default in older PostgreSQL versions.

REPLICA:
  Records enough for physical replication (exact byte-for-byte copy).
  Still does NOT record enough for logical decoding. Used for standby
  replicas, not CDC.

LOGICAL:
  Records full before and after images of changed rows. Includes enough
  information for logical decoding — exactly what CDC needs.
  Required for Debezium and any WAL-based CDC tool.
  Slight overhead: larger WAL files due to full row images.
  Setting: wal_level = logical  (requires PostgreSQL restart)

ADDITIONAL SETTINGS REQUIRED FOR CDC:
  max_replication_slots = 10  # how many logical replication consumers allowed
  max_wal_senders = 10        # parallel WAL streaming connections
  wal_sender_timeout = 60000  # ms — close idle WAL sender connections

HOW LOGICAL DECODING WORKS:
  PostgreSQL exposes two interfaces for logical decoding:
  1. pg_logical_slot_get_changes() — pull-based SQL function
  2. Streaming replication protocol — push-based (what Debezium uses)

  Debezium connects as a logical replication client over the replication
  protocol — exactly like a PostgreSQL standby would connect, but instead
  of applying the WAL to a replica database, it decodes it into events.

LOG SEQUENCE NUMBER (LSN):
  Every WAL record has an LSN — a monotonically increasing 64-bit integer
  representing its position in the WAL stream. Format: 0/1A3F2B8 (hex).
  Used for: tracking how far the consumer has read (confirmed_flush_lsn in
  the replication slot), ordering events (lower LSN = happened earlier),
  and resuming after failure (start reading from last confirmed LSN).

  Viewing current LSN and consumer lag:
    SELECT pg_current_wal_lsn();         -- where we are now
    SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes
    FROM pg_replication_slots
    WHERE slot_name = 'debezium_slot';   -- how far behind the consumer is

APPLICATION EXECUTES:
  BEGIN;
  UPDATE orders SET status = 'delivered', delivered_at = NOW()
  WHERE order_id = 9284751;
  COMMIT;

WAL RECORDS WRITTEN (binary, simplified representation):
  Record 1: XLOG_HEAP_UPDATE
    relation:    16423 (orders table OID)
    old_tuple:   [9284751, 'confirmed', NULL]
    new_tuple:   [9284751, 'delivered', '2026-03-17 20:14:32 UTC']
    lsn:         0/1A3F2B8
    xid:         847291             -- transaction ID

  Record 2: XLOG_XACT_COMMIT
    xid:         847291
    commit_time: 2026-03-17 20:14:32.847 UTC
    lsn:         0/1A3F2C4

1. Debezium reads binary WAL records via the replication protocol
2. Applies the 'pgoutput' logical replication plugin (built into PostgreSQL)
3. pgoutput translates OIDs to table names, type-decodes raw bytes
4. Debezium wraps the decoded event in an envelope, published to Kafka:

{
  "payload": {
    "before": {"order_id": 9284751, "status": "confirmed", "delivered_at": null},
    "after":  {"order_id": 9284751, "status": "delivered",
               "delivered_at": 1710706472000000},  -- microseconds since epoch
    "source": {
      "connector": "postgresql", "name": "freshcart", "db": "freshcart_prod",
      "ts_ms": 1710706472847, "snapshot": "false",
      "sequence": "[\"0/1A3F2B4\",\"0/1A3F2B8\"]",  -- [lastCommittedLsn, lsn]
      "table": "orders", "txId": 847291, "lsn": 28434104
    },
    "op": "u",             -- u=update, c=create, d=delete, r=read/snapshot
    "ts_ms": 1710706472901  -- when Debezium processed this event
  }
}

KEY FIELDS FOR DATA ENGINEERS:
op:           the operation type
before:       row values BEFORE the change (null for inserts)
after:        row values AFTER the change (null for deletes)
source.lsn:   position in WAL — use for ordering and dedup
source.txId:  transaction ID — group multi-table ops from same txn
source.ts_ms: event time at source (commit time)
ts_ms:        processing time (when Debezium emitted to Kafka)

SOURCE DB          KAFKA CONNECT           KAFKA              CONSUMERS
─────────────────────────────────────────────────────────────────────────
PostgreSQL    ←──  Debezium           →   Topic:             Spark
(primary)          PostgreSQL             freshcart.cdc       Streaming
                   Connector              .public.orders
                   (Kafka Connect         Topic:             Python
Replication  ←──   Worker process)   →   freshcart.cdc      Consumer
Slot                                      .public.customers
                   Schema Registry   →   Topic:             Flink
                   (sidecar)              freshcart.cdc
                                          .public.payments

KAFKA CONNECT:
  - Distributed worker framework that runs connector plugins
  - Scales horizontally — multiple workers share connector load
  - Stores connector offsets (LSN position) in a Kafka topic
    (not in the source database — offset is in _connect-offsets topic)
  - REST API: POST/GET/DELETE connectors, check status, restart

DEBEZIUM POSTGRESQL CONNECTOR:
  - Connects to PostgreSQL as a logical replication client
  - Reads WAL events via streaming replication protocol
  - Decodes binary WAL records using the pgoutput plugin
  - Publishes one Kafka message per database row change
  - Kafka message key: primary key of the changed row (for partition routing)

KAFKA TOPIC NAMING:
  Default pattern: {server.name}.{schema}.{table}
  Example: freshcart.public.orders
  One topic per table (recommended) — allows independent consumer groups.

OFFSET STORAGE:
  Debezium stores its WAL position (LSN) in a Kafka topic, NOT only in the
  source PostgreSQL replication slot. Two records of position exist:
  1. confirmed_flush_lsn in the PostgreSQL replication slot
     (what PostgreSQL knows the consumer has processed)
  2. Kafka Connect offset (internal __connect-offsets topic)
     (what Kafka Connect tracks as the connector's position)
  On restart: Debezium resumes from the Kafka Connect offset, then tells
  the PostgreSQL slot to start streaming from that LSN.

DELIVERY GUARANTEE:
  At-least-once: Debezium commits to Kafka before advancing the replication
  slot. On crash-restart, Debezium may re-emit the last batch of events
  (since the Kafka offset was not yet confirmed). Consumers MUST handle
  duplicates idempotently.

{
  "name": "freshcart-orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

    // Source database connection:
    "database.hostname":    "postgres-primary.internal",
    "database.port":        "5432",
    "database.user":        "debezium",
    "database.password":    "${file:/secrets/debezium.properties:db.password}",
    "database.dbname":      "freshcart_prod",
    "database.server.name": "freshcart",    // prefix for Kafka topic names

    // Table selection (include/exclude):
    "table.include.list": "public.orders,public.customers,public.payments",
    // OR exclude system tables:
    "table.exclude.list": "public.schema_migrations,public.sessions",

    // Column filtering (exclude PII from specific tables):
    "column.exclude.list": "public.customers.raw_phone,public.customers.ssn"
    // Excluded columns appear as null in CDC events — use for PII fields
    // that should not flow through Kafka
  }
}

The "${file:...}" syntax above is genuine Kafka Connect config-provider
syntax — it resolves the password from a local properties file at runtime,
not a template bug. This is one of the few places in this module where
"${" is correct as written.

"plugin.name":        "pgoutput",    // built into PostgreSQL 10+
"slot.name":          "debezium_freshcart",
"publication.name":   "dbz_freshcart_pub",

"snapshot.mode":           "initial",       // initial | never | schema_only | always
"snapshot.isolation.mode": "serializable",  // consistent snapshot

"topic.prefix": "freshcart.cdc",
// Results in topics: freshcart.cdc.public.orders, freshcart.cdc.public.customers

// Use Avro + Schema Registry in production, not raw JSON:
"key.converter":   "io.confluent.kafka.serializers.KafkaAvroSerializer",
"value.converter": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"key.converter.schema.registry.url":   "http://schema-registry:8081",
"value.converter.schema.registry.url": "http://schema-registry:8081"

"heartbeat.interval.ms": "30000",  // emit heartbeat every 30 seconds
// Without heartbeat: on low-write tables, the replication slot LSN never
// advances, WAL accumulates, slot lag appears to grow forever. Heartbeat
// emits a WAL record every N ms to advance the confirmed LSN.

"transforms":             "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.add.fields":           "op,table,lsn,source.ts_ms",
"transforms.unwrap.delete.handling.mode": "rewrite",
// ExtractNewRecordState flattens the envelope:
// BEFORE: {before: {...}, after: {...}, op: "u", source: {...}}
// AFTER:  {order_id: 9284751, status: "delivered", __op: "u", __lsn: ...}
// Simpler for consumers, but loses the before image

"tombstones.on.delete": "true",
// After a DELETE event, Debezium emits a tombstone (null-value message)
// with the same key. Kafka uses tombstones to compact deleted records.
// Consumers must handle null value messages without crashing.

"decimal.handling.mode": "string",
// Options: precise (Avro Decimal), double (lossy), string (safe for all consumers)
// Use "string" unless consumers can handle the Avro Decimal type correctly

"interval.handling.mode": "string"

PRODUCER (Debezium)                    SCHEMA REGISTRY         CONSUMER (Spark/Python)
──────────────────────────────────────────────────────────────────────────────────────
Detects orders table has:
  order_id: INT8, status: VARCHAR, amount: DECIMAL

Registers Avro schema → POST /subjects/freshcart.cdc.public.orders-value/versions
Receives schema_id: 42

Serializes message:                    Stores schema:
[magic_byte=0][schema_id=42][avro..]   ID 42 → Avro schema for orders v1
→ Publishes to Kafka topic

                                                                Consumer reads message:
                                                                Sees [schema_id=42][...]
                                                                Fetches schema 42 from
                                                                registry, deserializes
                                                                → {order_id: 9284751,
                                                                   status: "delivered",
                                                                   amount: 380.00}

Source DBA adds: ALTER TABLE orders ADD COLUMN delivery_fee DECIMAL(6,2) DEFAULT 0;

Debezium detects the schema change from the WAL DDL record.
New schema registers as version 2 (schema_id: 43):
  Added field: {"name": "delivery_fee", "type": ["null", "float"], "default": null}

COMPATIBILITY CHECK:
  Registry checks: is schema v2 backward-compatible with v1?
  Rule: adding a field with a default value is backward-compatible. ✓
  Rule: removing a required field is NOT backward-compatible. ✗

CONSUMER BEHAVIOUR:
  Old consumers (schema v1) reading a v2 message: delivery_fee is unknown → ignored ✓
  New consumers (schema v2) reading a v1 message: delivery_fee missing → default null ✓
  Both consumers continue working without redeployment.

BACKWARD (default):
  New schema can read data produced by the previous schema.
  Allows: adding fields with defaults, removing fields without defaults.
  Prevents: adding required fields, changing types.
  Best for: consumers that need to read old messages.

FORWARD:
  Previous schema can read data produced by the new schema.
  Opposite of backward — protects old consumers from new producers.

FULL:
  Both backward and forward. Most restrictive. Only additive changes.
  Best for: strict production environments.

NONE:
  No compatibility checking. Any change allowed. Use only in development.

from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError

# AvroConsumer handles schema fetching and Avro deserialization automatically:
consumer = AvroConsumer({
    'bootstrap.servers':   'kafka:9092',
    'group.id':            'freshcart-cdc-silver-writer',
    'schema.registry.url': 'http://schema-registry:8081',
    'auto.offset.reset':   'earliest',
    'enable.auto.commit':  False,   # manual commit for at-least-once
})
consumer.subscribe(['freshcart.cdc.public.orders', 'freshcart.cdc.public.payments'])

def process_event(msg) -> None:
    """Process one CDC event with schema-aware deserialization."""
    if msg.value() is None:
        # Tombstone event (after a delete) — value is null, key has the primary key
        key = msg.key()
        handle_tombstone(key['order_id'] if key else None)
        return

    event = msg.value()   # AvroConsumer deserialized using Schema Registry
    op = event.get('__op') or event.get('op')
    if op in ('c', 'r', 'u'):
        upsert_to_silver(event)
    elif op == 'd':
        mark_deleted_in_silver(event.get('order_id'))

while True:
    try:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        process_event(msg)
        consumer.commit()   # commit AFTER successful processing

    except SerializerError as e:
        # Schema deserialization failed — schema incompatibility
        print(f'Schema error: {e}')
        # Do NOT commit — message will be redelivered
        # Alert: schema compatibility issue needs investigation

NAIVE APPROACH (incorrect):
  BEGIN;
  UPDATE payments SET status = 'captured', captured_at = NOW()
  WHERE payment_id = 'pay_xxx';
  COMMIT;                                     -- Step 1: DB write succeeds
  producer.send('payments.captured', event)   -- Step 2: Kafka publish

FAILURE MODES:
  A) DB write succeeds, process crashes before Kafka publish
     → downstream services never notified — inconsistency
  B) Kafka publish succeeds, DB write fails (rolled back)
     → downstream services fulfill an order that was never paid for

TWO-PHASE COMMIT (XA)? Theoretically solves this, but Kafka does not
participate in XA transactions and XA is too slow for high-throughput
production systems — not practical here.

-- STEP 1: Create an outbox table (in the application database):
CREATE TABLE outbox_events (
    event_id        UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type  VARCHAR(50) NOT NULL,   -- 'order', 'payment', etc.
    aggregate_id    VARCHAR(50) NOT NULL,   -- the entity's ID
    event_type      VARCHAR(100) NOT NULL,  -- 'PaymentCaptured', etc.
    payload         JSONB       NOT NULL,   -- event data
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_outbox_created ON outbox_events(created_at);

-- STEP 2: Application writes to both payments AND outbox atomically:
BEGIN;
UPDATE payments SET status = 'captured', captured_at = NOW()
WHERE payment_id = 'pay_xxx';

INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload)
VALUES (
    'payment', 'pay_xxx', 'PaymentCaptured',
    '{"payment_id": "pay_xxx", "amount": 38000, "currency": "INR",
      "merchant_id": "merch_001", "captured_at": "2026-03-17T20:14:32Z"}'
);
COMMIT;
-- Both succeed or both fail — no inconsistency possible

{
  "transforms": "outbox",
  "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
  "transforms.outbox.table.field.event.id":      "event_id",
  "transforms.outbox.table.field.event.key":     "aggregate_id",
  "transforms.outbox.table.field.event.payload": "payload",
  "transforms.outbox.route.by.field":            "aggregate_type",
  "transforms.outbox.route.topic.replacement":   "outbox.${routedByValue}"
}
// Routes events to Kafka topics by aggregate_type:
// 'payment' events → Kafka topic: outbox.payment
// 'order' events   → Kafka topic: outbox.order
// The "${routedByValue}" above is genuine Debezium SMT placeholder syntax —
// not a template bug.

STEP 4: Consumers process events from outbox Kafka topics.

STEP 5 (optional): Delete processed outbox rows periodically:
  DELETE FROM outbox_events WHERE created_at < NOW() - INTERVAL '7 days';
  -- Or use a separate cleanup job — the outbox is not an event store,
  -- Kafka is. Outbox rows are only needed until CDC reads them.

WHY THIS WORKS:
  The outbox INSERT is in the same transaction as the business logic update.
  If the transaction commits: both the payment update AND the outbox event
  exist in the DB. If it rolls back: neither exists.
  CDC reads the outbox and publishes to Kafka — CDC provides at-least-once
  delivery. Downstream consumers handle duplicates with idempotency keys
  (event_id). Result: exactly-once business semantics from at-least-once
  delivery + idempotency.

WHAT CDC GUARANTEES:
  Within a single Kafka partition → events are strictly ordered by LSN.
  For a given primary key → all events for that row go to the same partition
  (because Kafka partition key = row's primary key by default).
  Therefore: all changes to order_id = 9284751 are ordered correctly.

WHAT CDC DOES NOT GUARANTEE:

1. ORDERING ACROSS TABLES:
   Event A (orders, LSN 1000) and Event B (customers, LSN 1001) may land in
   different Kafka partitions and be consumed out of order if the consumers
   for those topics run at different speeds. Example: a payment is captured
   (payments topic, LSN 5000) and the order status updated (orders topic,
   LSN 5001) — a consumer reading both topics may process the order update
   first, depending on consumer lag per topic.

2. ORDERING ACROSS KAFKA PARTITIONS:
   A high-write table may have multiple Kafka partitions. Events for
   different primary keys go to different partitions — ordered within each
   partition, but not between them. order_id 9284751 (partition 0) and
   order_id 9284752 (partition 1) events may arrive interleaved in any order.

3. SNAPSHOT + STREAM ORDERING:
   During initial snapshot, Debezium emits all existing rows as op=r events.
   Streaming changes begin from the snapshot LSN. The snapshot may take
   hours, and the source table is being modified during it — between
   snapshot completion and stream catchup, the consumer sees a mix of
   snapshot rows (as of snapshot start time) and stream changes (from
   snapshot start LSN forward). Upsert semantics reconcile both correctly.

SAFE PATTERN — process one table at a time, in order:
  consumer reads freshcart.cdc.public.orders one partition at a time
  → events for the same order are ordered, downstream safe

UNSAFE PATTERN — join across CDC streams in the consumer:
  consumer reads orders AND payments topics simultaneously, tries to join
  "when payment captured, also update order status" → ordering not
  guaranteed across topics → race condition possible

CORRECT PATTERN for cross-table consistency:
  Let each CDC stream write to its own Silver table independently. Let dbt
  handle the join in a SQL model — SQL joins are order-independent. Never
  implement cross-table joins in the CDC consumer layer.

SOURCE TRANSACTION:
  BEGIN;
  INSERT INTO orders (order_id, status) VALUES (9284753, 'placed');
  INSERT INTO order_items (order_id, product_id, qty) VALUES (9284753, 42, 2);
  COMMIT;                          -- both rows committed atomically

CDC EVENTS (Debezium emits, both sharing txId: 847292):
  {op: "c", table: "orders",      txId: 847292, ...}
  {op: "c", table: "order_items", txId: 847292, ...}
  -- may be on different Kafka topics/partitions

SOLUTION: enable Debezium's transaction metadata support:
  {"provide.transaction.metadata": "true"}
  Debezium emits BEGIN {txId, event_count} and COMMIT {txId, event_count,
  data_collections} events on a transaction metadata topic. A consumer can
  buffer events until the full transaction is received, then process
  atomically together.

SIMPLER APPROACH for analytics: don't try to preserve transaction
boundaries at the consumer layer. Write each event to its own Silver table
with upsert semantics, and run dbt models that JOIN across Silver tables —
dbt sees a consistent snapshot of all Silver tables at query time. This is
usually sufficient for analytical use cases.

Source: ALTER TABLE orders ADD COLUMN delivery_fee DECIMAL(6,2) DEFAULT 0;

Debezium behaviour:
  - Detects DDL change from WAL, registers a new Avro schema
  - New schema version is backward-compatible — Registry accepts it

Messages BEFORE the DDL: delivery_fee absent, consumers use default null
Messages AFTER the DDL:  delivery_fee present with value

Consumer impact: old consumers (schema v1) see null (backward-compatible) ✓
new consumers (schema v2) see the correct value ✓ — NO crash, NO redeploy.

Data warehouse impact:
  - dbt Silver model needs updating to SELECT delivery_fee
  - ALTER TABLE silver.orders ADD COLUMN delivery_fee DECIMAL(6,2);
  - Or: dbt run --full-refresh silver.orders

Source: ALTER TABLE orders RENAME COLUMN order_amount TO amount;

Debezium behaviour:
  - Removing "order_amount" WITHOUT a default: NOT backward-compatible!
  - Schema Registry REJECTS this schema if compatibility mode = BACKWARD
  - If blocked: connector pauses, requires manual intervention

SAFE MIGRATION APPROACH (avoid breaking changes):
1. Add the NEW column: ALTER TABLE orders ADD COLUMN amount DECIMAL(10,2);
2. Write to both columns temporarily (application change)
3. Backfill: UPDATE orders SET amount = order_amount WHERE amount IS NULL;
4. Consumer updated to read "amount" column
5. Drop old column: ALTER TABLE orders DROP COLUMN order_amount;
6. Update Debezium schema: new version removes order_amount
This staged migration takes longer but never breaks the pipeline.

Source: ALTER TABLE orders ALTER COLUMN status TYPE VARCHAR(50);  -- was VARCHAR(20)

Widening (VARCHAR(20) → VARCHAR(50)) is usually safe — values that were
valid VARCHAR(20) are still valid VARCHAR(50). Debezium emits strings, so
the width change is transparent.

Narrowing (VARCHAR(50) → VARCHAR(20)) is dangerous — existing data may
violate the new constraint. Debezium/Schema Registry may reject this if
truncation is detected.

DEBEZIUM SCHEMA CHANGE EVENT: on every DDL, Debezium emits an event to a
separate topic ({server.name}.schema-changes.{database}) containing the
raw DDL text and the full new column list. Useful for auditing schema
changes and alerting when unexpected DDL occurs.

PostgreSQL → Debezium → Kafka → [Spark Streaming / Flink] → Delta Lake
                                  (micro-batch consumer)      (Bronze layer)

At 10,000 events/second, writing each event as its own file means 10,000
S3 PUT requests/second, each Parquet file ~1 KB. After 1 hour: 36,000,000
tiny files — unusable for analytics.

CORRECT APPROACH: buffer in Kafka, write batches to Delta Lake on a
micro-batch trigger.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import StructType, LongType, StringType

spark = SparkSession.builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "freshcart.cdc.public.orders") \
    .option("startingOffsets", "latest") \
    .option("maxOffsetsPerTrigger", 500_000) \
    .load()

orders_schema = StructType() \
    .add("order_id",   LongType()) \
    .add("status",     StringType()) \
    .add("amount",     StringType()) \  # decimal as string
    .add("updated_at", LongType()) \    # microseconds epoch
    .add("__op",       StringType()) \  # operation type
    .add("__lsn",      LongType())       # WAL LSN for ordering

parsed = raw.select(from_json(col("value").cast("string"), orders_schema).alias("e")) \
    .select("e.*").withColumn("ingested_at", current_timestamp())

def upsert_to_delta(batch_df, batch_id):
    from delta.tables import DeltaTable
    if DeltaTable.isDeltaTable(spark, "s3://freshcart-lake/bronze/orders"):
        delta_table = DeltaTable.forPath(spark, "s3://freshcart-lake/bronze/orders")
        delta_table.alias("target").merge(
            batch_df.alias("source"), "target.order_id = source.order_id"
        ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
    else:
        batch_df.write.format("delta").mode("overwrite") \
            .partitionBy("date").save("s3://freshcart-lake/bronze/orders")

query = parsed.writeStream \
    .foreachBatch(upsert_to_delta) \
    .option("checkpointLocation", "s3://freshcart-lake/checkpoints/orders_cdc") \
    .trigger(processingTime="5 minutes") \
    .start()
query.awaitTermination()

HANDLING DELETE EVENTS IN THE DATA LAKE:
  Option A — soft-delete flag: store __op/__deleted in Delta Lake, dbt
    Silver filters WHERE __op != 'd'. Preserves full history for audit.
  Option B — hard delete via Delta MERGE DELETE clause:
    .whenMatchedDelete(condition="source.__op = 'd'"). Cleaner, loses history.
  Option C — separate deleted-records table: route deletes to
    bronze.orders_deletes, dbt Silver EXCEPTs those IDs. Keeps both views.

COMPACTION SCHEDULE:
  CDC streams create many small Delta files per 5-minute trigger. Run
  OPTIMIZE daily:
    OPTIMIZE delta.`s3://freshcart-lake/bronze/orders`
    WHERE date >= current_date - 7;

-- Replication slot lag:
SELECT slot_name, active,
    pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes,
    pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) / 1024 / 1024 AS lag_mb,
    now() - pg_last_xact_replay_timestamp() AS replication_delay
FROM pg_replication_slots;

-- ALERT thresholds:
-- lag_mb > 1024 (1 GB): WARNING — consumer is falling behind
-- lag_mb > 10240 (10 GB): CRITICAL — risk of disk fill, investigate immediately
-- active = false: CRITICAL — slot exists but consumer is disconnected

-- WAL disk usage:
SELECT pg_size_pretty(sum(size)) AS wal_disk_usage FROM pg_ls_waldir();

import requests

# Check connector status:
response = requests.get('http://kafka-connect:8083/connectors/freshcart-orders-cdc/status')
status = response.json()
# Expected: {"connector": {"state": "RUNNING"}, "tasks": [{"state": "RUNNING", ...}]}
# Alert if state != "RUNNING"

# Pause/resume connector (for maintenance):
requests.put('http://kafka-connect:8083/connectors/freshcart-orders-cdc/pause')
requests.put('http://kafka-connect:8083/connectors/freshcart-orders-cdc/resume')

# Restart a failed task:
requests.post('http://kafka-connect:8083/connectors/freshcart-orders-cdc/tasks/0/restart')

# kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
#   --describe --group freshcart-cdc-silver-writer
#
# TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# freshcart.cdc.public.orders    0          1847291         1847301         10
# freshcart.cdc.public.orders    1          2039847         2039847         0

from confluent_kafka.admin import AdminClient
admin = AdminClient({'bootstrap.servers': 'kafka:9092'})
offsets = admin.list_consumer_group_offsets(['freshcart-cdc-silver-writer'])
# ALERT if any partition lag > 10,000 events or > 5 minutes of events

import psycopg2, requests

def check_cdc_health(pg_conn_str: str, kafka_connect_url: str) -> list[str]:
    """Check CDC health, return list of alert messages."""
    alerts = []

    with psycopg2.connect(pg_conn_str) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT slot_name, active,
                       pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) / 1024 / 1024
                FROM pg_replication_slots
            """)
            for slot_name, active, lag_mb in cur.fetchall():
                if not active:
                    alerts.append(f'CRITICAL: Slot {slot_name} is INACTIVE')
                elif lag_mb and lag_mb > 10240:
                    alerts.append(f'CRITICAL: Slot {slot_name} lag = {lag_mb:.0f} MB')
                elif lag_mb and lag_mb > 1024:
                    alerts.append(f'WARNING: Slot {slot_name} lag = {lag_mb:.0f} MB')

    resp = requests.get(f'{kafka_connect_url}/connectors?expand=status', timeout=10)
    for name, info in resp.json().items():
        state = info['status']['connector']['state']
        if state != 'RUNNING':
            alerts.append(f'CRITICAL: Connector {name} state = {state}')
        for task in info['status']['tasks']:
            if task['state'] != 'RUNNING':
                alerts.append(f'CRITICAL: Connector {name} task {task["id"]} = {task["state"]}')

    return alerts

FAILURE 1: Connector FAILED state
  Likely cause: PostgreSQL connectivity issue, schema change, credential expiry
  Recovery: (1) GET /connectors/{name}/status → error_trace shows the exact
  exception. (2) Fix the root cause. (3) Restart the failed task:
  POST /connectors/{name}/tasks/0/restart. (4) If it keeps failing: DELETE
  and recreate the connector (may require re-snapshot).

FAILURE 2: Replication slot lag growing (consumer slow)
  Risk: if lag grows indefinitely, source disk fills.
  Recovery: (1) identify the bottleneck — consumer or broker slow? Check
  kafka-consumer-groups.sh --describe. (2) if consumer slow: scale up the
  consumer group. (3) if Kafka is the bottleneck: increase topic partitions.
  (4) if unrecoverable and lag > 50 GB with no improvement in 30 min:
  consider dropping the slot (accepts data loss) to protect the source DB.

FAILURE 3: Consumer offset behind Kafka retention
  Symptom: "Offset 0 is not available, earliest is 48000000" — consumer was
  paused longer than Kafka's retention period.
  Recovery: reset consumer group offset to earliest available, then either
  re-snapshot from current state or reprocess everything (upserts handle it).

FAILURE 4: PostgreSQL primary failover
  Symptom: Debezium loses connection to the old primary.
  Recovery: update database.hostname to the new primary (ideally via a DNS
  alias), restart the connector — it resumes from the last confirmed LSN.
  Note: WAL LSN sequence resets if the replica wasn't fully caught up —
  some events may be replayed or lost; heartbeats help detect this.

FAILURE 5: Schema change breaks the consumer
  Symptom: SerializerError or NullPointerException on new events.
  Recovery: check GET /subjects/{topic}-value/versions/latest. If the
  schema was accepted, update consumer code for the new field. If rejected
  by the Registry (a genuine breaking change), work with the source team
  on a safe staged migration instead.

FreshCart’s legal team informs the data engineering team that under GDPR (applicable to users in the EU), customers can request deletion of their personal data. When a deletion request is processed by the application team, the customer row in PostgreSQL is hard-deleted. The data engineering team has 30 days to ensure that data is also removed from the analytics warehouse and all downstream systems.

The incremental ingestion pipeline currently running on the customers table uses an updated_at watermark and cannot detect hard deletes. Silver and Gold tables in Snowflake still contain data for deleted customers. This is a compliance violation. The solution: CDC on the customers table.

Step 1-2 — switch to CDC, handle delete events

STEP 1: Switch customers table ingestion from incremental to CDC.
  Configure Debezium to capture the customers table.
  Create a consumer that writes to bronze.customers (Delta Lake).
  Decommission the incremental Python pipeline.

STEP 2: Handle DELETE events in the consumer:
  def process_customer_event(event: dict) -> None:
      op = event.get('__op') or event.get('op')
      if op == 'd':
          customer_id = event.get('customer_id') or \
                        (event.get('before') or {}).get('customer_id')
          spark.sql(f"""
              DELETE FROM delta.`s3://freshcart-lake/bronze/customers`
              WHERE customer_id = {customer_id}
          """)
          log_erasure(customer_id, reason='gdpr_deletion',
                      erased_at=datetime.now(timezone.utc))
          publish_to_kafka('customer.erasure', {
              'customer_id': customer_id,
              'erased_at':   datetime.now(timezone.utc).isoformat(),
          })
      elif op in ('c', 'u', 'r'):
          upsert_to_bronze(event)

Step 3-4 — exclude erased customers, cascade to Gold

-- models/silver/customers.sql
WITH bronze AS (
    SELECT * FROM {{ ref('bronze_customers') }}
),
erased AS (
    SELECT customer_id FROM {{ ref('erasure_log') }}
    WHERE erased_at >= DATEADD('day', -30, CURRENT_DATE)
)
SELECT b.*
FROM bronze b
LEFT JOIN erased e USING (customer_id)
WHERE e.customer_id IS NULL   -- exclude erased customers

-- STEP 4: cascade to Gold
-- dbt run --select +customers+  (rebuilds all downstream models)

Step 5-6 — purge Kafka PII, verify compliance

STEP 5: Purge PII from Kafka.
  Kafka retains messages for the configured retention period, and CDC
  events for a deleted customer contain PII in the 'before' field.
  Use Kafka's tombstone mechanism: a null-value message with the same key
  (customer_id) triggers log compaction, removing all older messages with
  that key once compaction runs.

STEP 6: Verify compliance.
  SELECT COUNT(*) FROM silver.customers WHERE customer_id = {deleted_id};
  -- Must return 0
  SELECT COUNT(*) FROM gold.customer_ltv WHERE customer_id = {deleted_id};
  -- Must return 0

Result: GDPR erasure complete within minutes of the application deletion,
well within the 30-day compliance window.

This pattern — CDC enabling reliable hard-delete propagation — is one of the most common business-driven reasons for adopting CDC beyond analytics performance. Incremental ingestion simply cannot support right-to-erasure requirements. CDC is the only alternative to full periodic reloads for propagating deletions.