Failure Scenario Drills

Learning Goals

• Observe consumer lag metrics when downstream systems are unavailable
• Understand how Kafka buffers changes during sink outages
• Identify at what point lag becomes critical
• Practice monitoring and alerting thresholds

Setup

1. Baseline metrics: Record current consumer lag and throughput

   copykafka-consumer-groups --bootstrap-server localhost:9092 \
     --describe --group sink-group-inventory

2. Enable DLQ (optional but recommended): Configure your sink connector with DLQ settings

Execute

1. Stop the sink connector:

   copy# If using JDBC Sink
   curl -X DELETE http://localhost:8083/connectors/jdbc-sink-connector
   
   # Or pause it
   curl -X PUT http://localhost:8083/connectors/jdbc-sink-connector/pause

2. Generate source changes: Insert, update, or delete rows in source database

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "INSERT INTO public.app_customer (name, email) VALUES ('Test User', 'test@example.com');"

3. Watch lag grow: Monitor consumer group lag every 30 seconds

   copy# Watch lag in real-time (Ctrl+C to stop)
   watch -n 5 "kafka-consumer-groups --bootstrap-server localhost:9092 \
     --describe --group sink-group-inventory | grep -v 'Consumer group'"

Observe & Document

Recovery

copy# Resume or recreate the sink connector
curl -X PUT http://localhost:8083/connectors/jdbc-sink-connector/resume

# Or re-register it
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d @sink-config.json

Validation Checklist

• Lag returns to zero within expected time
• All accumulated changes applied to sink
• No duplicate rows in sink (if using idempotent writes)
• Monitoring alerts fired when lag exceeded threshold

Learning Goals

• Trigger and identify serialization errors in the pipeline
• Configure and monitor Dead Letter Queue (DLQ) topics
• Practice DLQ triage and message recovery
• Understand error tolerance policies

Setup

1. Enable DLQ on sink connector: Update connector config

   copy{
     "name": "jdbc-sink-connector",
     "config": {
       "errors.tolerance": "all",
       "errors.deadletterqueue.topic.name": "dlq.sink.inventory",
       "errors.deadletterqueue.context.headers.enable": "true",
       "errors.log.enable": "true",
       "errors.log.include.messages": "true"
     }
   }

2. Create a consumer for DLQ monitoring:

   copykafka-console-consumer --bootstrap-server localhost:9092 \
     --topic dlq.sink.inventory --from-beginning \
     --property print.headers=true \
     --property print.timestamp=true

Execute

1. Method 1: Inject incompatible data type

   copy# Assuming sink expects integer but source sends string
   docker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "INSERT INTO public.problematic_table (id, age) VALUES (999, 'not-a-number');"

2. Method 2: Create missing target table scenario

   copy# Rename target table to simulate missing destination
   # Note: postgres-sink is the sink database container (port 5433)
   docker exec -it postgres-sink psql -U sink_user -d warehouse -c \
     "ALTER TABLE public.app_customer RENAME TO app_customer_backup;"

3. Method 3: Insert oversized payload

   copy# Insert a very large text field that exceeds sink column limit
   # (Assuming sink has VARCHAR(255) limit and this creates a 5000 char string)
   docker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "INSERT INTO public.app_customer (name, email) VALUES (repeat('X', 5000), 'large@example.com');"

Observe & Triage

1. Check connector status:

   copycurl -s http://localhost:8083/connectors/jdbc-sink-connector/status | jq '.tasks[].trace'

2. Inspect DLQ messages:

   copy# List messages with error context
   kafka-console-consumer --bootstrap-server localhost:9092 \
     --topic dlq.sink.inventory --from-beginning \
     --property print.headers=true \
     --max-messages 5

3. Identify error pattern: Look for headers like:
   • __connect.errors.topic — Original topic
   • __connect.errors.partition — Partition number
   • __connect.errors.offset — Original offset
   • __connect.errors.exception.class.name — Error type
   • __connect.errors.exception.message — Error details

Recovery Strategies

copy# Use kafka-console-producer with headers
# Or use a DLQ replay tool

Validation Checklist

• DLQ captured poison pill messages
• Connector remained running (didn't crash-loop)
• Error headers contain actionable diagnostics
• Good messages continued processing normally
• DLQ alerts fired when messages arrived

Learning Goals

• Understand how schema changes propagate through CDC pipelines
• Identify breaking vs. non-breaking schema changes
• Practice schema evolution strategies
• Test schema registry integration (if applicable)

Setup

1. Document current schema:

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "\d public.app_customer"

2. Baseline a known-good record:

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "SELECT * FROM public.app_customer LIMIT 1;"

Execute: Breaking Changes

1. Scenario A: Drop a column with active data

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "ALTER TABLE public.app_customer DROP COLUMN email;"

   Impact: Sink may fail if it expects email column; Debezium emits event without field

2. Scenario B: Change column type (incompatible)

   copy# Change integer to text (potentially lossy)
   docker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "ALTER TABLE public.app_customer ALTER COLUMN id TYPE TEXT USING id::TEXT;"

   Impact: Type mismatch may cause sink errors if target schema is strict

3. Scenario C: Add NOT NULL constraint on existing nullable column

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "ALTER TABLE public.app_customer ADD COLUMN phone VARCHAR(20) NOT NULL DEFAULT '';"

   Impact: Existing rows get default; new inserts must provide value

Execute: Compatible Changes

1. Scenario D: Add nullable column (forward-compatible)

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "ALTER TABLE public.app_customer ADD COLUMN loyalty_points INTEGER DEFAULT 0;"

   ✓ Safe: Sink ignores unknown columns or auto-adds if using schema sync

2. Scenario E: Increase column width

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "ALTER TABLE public.app_customer ALTER COLUMN name TYPE VARCHAR(500);"

   ✓ Safe: Widening columns rarely breaks downstream systems

Observe Behavior

Recovery Patterns

Validation Checklist

• Documented which changes are safe vs. breaking
• Tested sink resilience to unknown fields
• Verified DLQ captures schema mismatches
• Confirmed schema registry versioning (if using)
• Created runbook for schema change approval process

Learning Goals

• Understand how CDC connectors manage offsets and state
• Observe snapshot vs. streaming mode behavior
• Practice safe offset resets and replays
• Validate idempotency of sink writes during replays

Setup

1. Ensure idempotent sink: Verify sink uses UPSERT/MERGE, not INSERT

   copy# Check sink connector config
   curl -s http://localhost:8083/connectors/jdbc-sink-connector/config | \
     jq '.["insert.mode"]'  # should be "upsert"

2. Backup current state:

   copy# Export connector offsets (Kafka Connect stores in internal topic)
   kafka-console-consumer --bootstrap-server localhost:9092 \
     --topic my_connect_offsets --from-beginning --timeout-ms 5000 > offsets-backup.json

3. Record sink row count:

   copy# Note: postgres-sink is the sink database container (port 5433)
   docker exec -it postgres-sink psql -U sink_user -d warehouse -c \
     "SELECT COUNT(*) FROM public.app_customer;"

Execute: Offset Reset

1. Stop connector gracefully:

   copycurl -X DELETE http://localhost:8083/connectors/inventory-connector
   curl -X DELETE http://localhost:8083/connectors/jdbc-sink-connector

2. Wipe connector offsets:

   copy# Delete Postgres replication slot (forces new snapshot)
   docker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "SELECT pg_drop_replication_slot('debezium');"
   
   # Delete Connect offsets for this connector
   # (In production, use offset management APIs; for lab, recreate with new name)

3. Re-register source connector with fresh snapshot:

   copycurl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
     "name": "inventory-connector-v2",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
       "snapshot.mode": "initial",
       "database.hostname": "postgres",
       "database.port": "5432",
       "database.user": "start_data_engineer",
       "database.password": "password",
       "database.dbname": "inventory",
       "topic.prefix": "server2",
       "table.include.list": "public.app_customer",
       "slot.name": "debezium_v2"
     }
   }'

4. Re-register sink connector to consume from new topic:

   copycurl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
     "name": "jdbc-sink-connector-v2",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
       "tasks.max": "1",
       "topics": "server2.public.app_customer",
       "connection.url": "jdbc:postgresql://postgres-sink:5432/warehouse",
       "connection.user": "sink_user",
       "connection.password": "password",
       "auto.create": "false",
       "auto.evolve": "true",
       "insert.mode": "upsert",
       "pk.mode": "record_key",
       "pk.fields": "id",
       "table.name.format": "app_customer",
       "transforms": "unwrap",
       "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
       "transforms.unwrap.drop.tombstones": "false",
       "transforms.unwrap.delete.handling.mode": "rewrite"
     }
   }'

   Note: The sink connector now consumes from server2.public.app_customer (matching the new source connector's topic). The insert.mode=upsert ensures idempotent writes during replay validation.

Observe Snapshot Phase

1. Watch connector task state:

   copywatch -n 2 "curl -s http://localhost:8083/connectors/inventory-connector-v2/status | jq '.tasks[0].state'"

2. Monitor snapshot progress:

   copy# Check connector logs for snapshot messages
   docker logs -f connect 2>&1 | grep -i snapshot

3. Count messages during snapshot:

   copykafka-console-consumer --bootstrap-server localhost:9092 \
     --topic server2.public.app_customer --from-beginning \
     --timeout-ms 10000 | wc -l

Observe Streaming Phase

1. Make a live change:

   copydocker exec -it postgres psql -U start_data_engineer -d inventory -c \
     "UPDATE public.app_customer SET name = 'Updated After Snapshot' WHERE id = 1;"

2. Verify streaming event arrives:

   copykafka-console-consumer --bootstrap-server localhost:9092 \
     --topic server2.public.app_customer \
     --offset latest --partition 0 --timeout-ms 5000

   Look for "op":"u" in the JSON payload for update operation

Validate Idempotency

Duplicate Detection Query

copy-- Run on sink database
SELECT id, COUNT(*) AS occurrences
FROM public.app_customer
GROUP BY id
HAVING COUNT(*) > 1;

Expected: Zero rows (no duplicates)

Validation Checklist

• Connector completed snapshot phase (check logs)
• Streaming mode activated after snapshot
• Sink has no duplicate primary keys
• Latest updates preserved (not overwritten by snapshot data)
• Offset reset procedure documented in runbook