16. Cheat sheet

Everything worth keeping within reach.

The shape

   APP ──one transaction──▶ DATABASE
                              │  its own log
                              ▼
                          DEBEZIUM
                              │  change events
                              ▼
                       KAFKA / Pulsar / ...
                              │
                  ┌───────────┼───────────┐
                  ▼           ▼           ▼
              search      warehouse    cache

One task per connector. Always. It cannot be scaled out.

Source setup

PostgreSQL

wal_level = logical              # ← NEEDS A RESTART
max_replication_slots = 10
max_wal_senders = 10
max_slot_wal_keep_size = 64GB    # ← SET THIS. it saves the database.

CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD '...';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER TABLE orders REPLICA IDENTITY FULL;   -- ← for a real `before`

MySQL

log_bin = mysql-bin
binlog_format = ROW              # ← MANDATORY
binlog_row_image = FULL          # ← for a real `before`
gtid_mode = ON                   # ← survives failover
enforce_gtid_consistency = ON
binlog_expire_logs_seconds = 604800   # bounds your outage tolerance

GRANT SELECT, RELOAD, SHOW DATABASES,
      REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';

MongoDB / SQL Server

   MONGO    capture.mode=change_streams_update_full   ← or you get a patch
            pre-images enabled for a real `before`
   SQL SVR  EXEC sys.sp_cdc_enable_db;
            EXEC sys.sp_cdc_enable_table ...
            ⚠ SQL Server AGENT must be running

Connector config

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "database.hostname": "postgres-replica",
  "database.dbname": "app",
  "topic.prefix": "cdc",

  "plugin.name": "pgoutput",
  "slot.name": "debezium_orders",
  "publication.autocreate.mode": "filtered",

  "table.include.list": "public.orders,public.customers",
  "column.exclude.list": "public.customers.ssn",

  "heartbeat.interval.ms": "10000",
  "heartbeat.action.query": "INSERT INTO debezium_heartbeat (ts) VALUES (now())",

  "snapshot.mode": "initial",
  "signal.data.collection": "public.debezium_signal",
  "incremental.snapshot.chunk.size": "1024",

  "decimal.handling.mode": "string",
  "time.precision.mode": "connect",
  "tombstones.on.delete": "true",

  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://registry:8081",

  "producer.override.acks": "all",
  "producer.override.enable.idempotence": "true",
  "producer.override.compression.type": "zstd",
  "producer.override.linger.ms": "20",

  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "cdc-dlq",
  "errors.deadletterqueue.context.headers.enable": "true"
}

The change event

{ "before": {...} | null,
  "after":  {...} | null,
  "source": {"ts_ms": <DB COMMIT TIME>, "table":..., "lsn":...},
  "op": "c|u|d|r|t|m",
  "ts_ms": <WHEN DEBEZIUM SAW IT> }

   c  insert       u  update       d  delete
   r  SNAPSHOT READ — the row already existed.
      NOT a creation. ← the email-everyone bug
   t  truncate     m  logical message

   KEY = the primary key → the partition
       → ordering per row, compaction, upsert

   TOMBSTONE = same key, null value, after a
   delete. for compaction. consumers must
   expect it.

   source.ts_ms vs ts_ms  = PIPELINE LAG

   ⚠ `before` is only full if the DATABASE
     was configured for it (REPLICA IDENTITY
     FULL / binlog_row_image FULL)
   ⚠ decimal.handling.mode=precise → BASE64
     in JSON. use `string`.
   ⚠ temporal values are often MICROseconds,
     not milliseconds

Signals

CREATE TABLE debezium_signal (
  id VARCHAR(42) PRIMARY KEY, type VARCHAR(32), data VARCHAR(2048));

-- snapshot one table, streaming continues
INSERT INTO debezium_signal VALUES ('s1','execute-snapshot',
 '{"data-collections":["public.order_items"],"type":"INCREMENTAL"}');

-- a targeted backfill
INSERT INTO debezium_signal VALUES ('s2','execute-snapshot',
 '{"data-collections":["public.orders"],"type":"INCREMENTAL",
   "additional-conditions":[{"data-collection":"public.orders",
     "filter":"region = ''EMEA''"}]}');

INSERT INTO debezium_signal VALUES ('s3','pause-snapshot', NULL);
INSERT INTO debezium_signal VALUES ('s4','resume-snapshot', NULL);
INSERT INTO debezium_signal VALUES ('s5','stop-snapshot', NULL);

   snapshot.mode
     initial       snapshot then stream  ← default
     never         stream only
     initial_only  snapshot then stop
     no_data       schema only
     when_needed   ⚠ will re-snapshot 400M rows
                     unattended at 3am
     always        testing only

SMTs

transforms=unwrap,route

transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields=op,source.ts_ms,source.lsn
transforms.unwrap.add.headers=op
transforms.unwrap.delete.tombstone.handling.mode=rewrite
#                                                 ▲ `drop` leaves
#                                                   deleted rows in
#                                                   your sink FOREVER

transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
transforms.route.regex=cdc\\.public\\.(.*)
transforms.route.replacement=$1

   ✓ SMT      unwrap, rename, mask, route,
              add metadata, simple filter
   ✗ FLINK    join, aggregate, deduplicate,
              enrich, window, anything
              STATEFUL

   SMTs run in the ONE task. thin config =
   higher throughput and a smaller blast
   radius.

The outbox

CREATE TABLE outbox (
  id UUID PRIMARY KEY, aggregatetype VARCHAR(255),
  aggregateid VARCHAR(255), type VARCHAR(255), payload JSONB);

BEGIN;
  UPDATE orders SET status='shipped' WHERE id=4471;
  INSERT INTO outbox VALUES (gen_random_uuid(),'Order','4471',
    'OrderShipped','{"orderId":4471,...}');
COMMIT;

transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.route.by.field=aggregatetype
transforms.outbox.route.topic.replacement=outbox.event.${routedByValue}
transforms.outbox.table.field.event.key=aggregateid
transforms.outbox.table.field.event.payload=payload
transforms.outbox.table.fields.additional.placement=type:header:eventType
transforms.outbox.table.expand.json.payload=true

Raw CDC for data movement. Outbox for service integration.

Topics

topic.creation.enable=true
topic.creation.default.replication.factor=3
topic.creation.default.partitions=12
topic.creation.default.cleanup.policy=compact

   compact  → the topic IS the table. a new
              consumer bootstraps from the
              beginning with NO load on the
              source. ← the best property here.
   needs tombstones. never instantaneous.

   ⚠ PLAN PARTITION COUNT UP FRONT.
     adding partitions breaks ordering.

   ⚠ SCHEMA HISTORY TOPIC (MySQL/SQLSvr/Oracle)
     1 partition, RF 3, retention -1,
     NEVER COMPACT, NEVER EXPIRE.
     losing it = the connector will not start.

Diagnostics

-- POSTGRES: the query that matters most
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(
         pg_current_wal_lsn(), confirmed_flush_lsn)) AS retained_wal
FROM pg_replication_slots;

SELECT pg_drop_replication_slot('debezium_orders');  -- the 3am decision

# TASK state, not connector state
curl -s http://connect:8083/connectors/orders-cdc/status | jq -r '.tasks[].state'

curl -X POST http://connect:8083/connectors/orders-cdc/restart?includeTasks=true
curl -X PUT  http://connect:8083/connectors/orders-cdc/pause
curl -s http://connect:8083/connectors/orders-cdc/config | jq

   JMX METRICS
     MilliSecondsBehindSource   ← the headline
     QueueRemainingCapacity     ← backpressure
     Connected
     RowsScanned                ← snapshot progress
     SnapshotRunning

   ALERT ON, IN ORDER
   1  slot retained_wal / active=false  ← DATABASE side
   2  task state != RUNNING
   3  MilliSecondsBehindSource rising
   4  QueueRemainingCapacity near zero
   5  DLQ depth
   6  source-vs-sink row counts diverging

Troubleshooting

Symptom	Cause	Fix
DB disk filling	slot not advancing	`max_slot_wal_keep_size`; drop the slot
Stale sink, green dashboard	task FAILED, connector RUNNING	alert on task state
WAL grows, lag is zero	quiet captured table	heartbeats
“binlog no longer available”	outage > retention	re-snapshot; raise retention
“Could not find schema history”	topic compacted/expired	never compact it; recovery mode
`before` is just the key	REPLICA IDENTITY / row_image	change it at the DB
Sink has deleted rows	delete handling = drop	`rewrite`; reconcile counts
Duplicate emails	offset replay	dedupe on `source.lsn`; drive from state
Everyone got a welcome email	`r` treated as `c`	handle `r` explicitly
400M events at 2am	rewriting `ALTER TABLE`	know which ALTERs rewrite
Dates in year 57000	micros read as millis	`time.precision.mode`
Base64 amounts	`decimal.handling.mode=precise`	use `string`
Snapshot restarted from zero	initial snapshots don’t resume	incremental snapshots
Two connectors flapping	same slot / same server.id	unique names

The ten rules

Set max_slot_wal_keep_size. It turns a database outage into a re-snapshot.
Alert on the replication slot from the database side. It is the only alert that protects the database.
Alert on task state, not connector state.
Heartbeats on every filtered connector.
Every consumer must be idempotent, and must distinguish r from c and handle d.
Avro with a registry, not the JSON default — often 30× smaller.
Use incremental snapshots. Resumable, per-table, pausable.
Never compact or expire the schema history topic.
Reconcile source and sink counts on a schedule. It is the only check that catches silent divergence.
The DBA team must know this exists, be alerted, and have a runbook. Their absence is a veto, not an obstacle.

The guide

Written by Sepuri Sai Krishna