1
2
3
4

-- Required MySQL configuration
log_bin = ON
binlog_format = ROW
binlog_row_image = FULL    -- Captures before AND after images

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

{
  "schema": { ... },
  "payload": {
    "before": {                    // Row state BEFORE the change (null for INSERT)
      "id": 12345,
      "status": "pending",
      "amount": 9999,
      "updated_at": 1711497600000
    },
    "after": {                     // Row state AFTER the change (null for DELETE)
      "id": 12345,
      "status": "shipped",
      "amount": 9999,
      "updated_at": 1711584000000
    },
    "source": {
      "version": "2.5.0.Final",
      "connector": "postgresql",
      "name": "production-db",
      "ts_ms": 1711584000123,      // Transaction commit timestamp
      "snapshot": "false",
      "db": "myapp",
      "schema": "public",
      "table": "orders",
      "txId": 847234,              // PostgreSQL transaction ID
      "lsn": 24522983             // Log Sequence Number (WAL position)
    },
    "op": "u",                     // c=create, u=update, d=delete, r=read(snapshot)
    "ts_ms": 1711584000456,        // Time Debezium processed the event
    "transaction": {
      "id": "847234",
      "total_order": 3,
      "data_collection_order": 1
    }
  }
}

1
2
3
4
5
6

# Install Strimzi operator
helm repo add strimzi https://strimzi.io/charts/
helm install strimzi-operator strimzi/strimzi-kafka-operator \
  --namespace kafka \
  --create-namespace \
  --set watchAnyNamespace=true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# kafka-connect.yaml — Kafka Connect cluster with Debezium
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect
  namespace: kafka
  annotations:
    strimzi.io/use-connector-resources: "true"  # Manage connectors as CRDs
spec:
  version: 3.7.0
  replicas: 3
  bootstrapServers: kafka-cluster-kafka-bootstrap:9092
  config:
    group.id: debezium-connect-cluster
    offset.storage.topic: debezium-connect-offsets
    config.storage.topic: debezium-connect-configs
    status.storage.topic: debezium-connect-status
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
    # Key and value converters
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false

  # Build a custom image with the Debezium connector JARs
  build:
    output:
      type: docker
      image: myregistry.example.com/debezium-connect:latest
    plugins:
      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.5.0.Final/debezium-connector-postgres-2.5.0.Final-plugin.tar.gz
      - name: debezium-mysql-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.5.0.Final/debezium-connector-mysql-2.5.0.Final-plugin.tar.gz

  resources:
    requests:
      cpu: 500m
      memory: 1Gi
    limits:
      cpu: 2
      memory: 4Gi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# docker-compose.yml
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  postgres:
    image: postgres:16
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: secret
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"          # Required for CDC
      - "-c"
      - "max_replication_slots=4"
      - "-c"
      - "max_wal_senders=4"

  kafka-connect:
    image: debezium/connect:2.5
    depends_on: [kafka, postgres]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: debezium
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://kafka-connect:8083

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- Create a dedicated replication user (never use the superuser)
CREATE USER debezium WITH REPLICATION LOGIN PASSWORD 'secure-password';

-- Grant SELECT on tables to capture
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;

-- For Debezium to read the initial snapshot, also grant CONNECT
GRANT CONNECT ON DATABASE myapp TO debezium;

-- Create a publication for the tables you want to capture
-- (pgoutput plugin requires a publication)
CREATE PUBLICATION debezium_publication
  FOR TABLE public.orders, public.customers, public.products;

-- Verify WAL level
SHOW wal_level;  -- Must return 'logical'

1

# Using Strimzi KafkaConnector CRD (GitOps-friendly)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# postgres-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: postgres-orders-connector
  namespace: kafka
  labels:
    strimzi.io/cluster: debezium-connect
spec:
  class: io.debezium.connector.postgresql.PostgresConnector
  tasksMax: 1  # PostgreSQL connector is single-threaded (one slot per connector)
  config:
    # Connection
    database.hostname: postgres.databases.svc.cluster.local
    database.port: "5432"
    database.user: debezium
    database.password: ${file:/opt/kafka/external-configuration/debezium-secrets/password}
    database.dbname: myapp
    database.server.name: production-db   # Prefix for all topic names

    # What to capture
    plugin.name: pgoutput
    publication.name: debezium_publication
    table.include.list: public.orders,public.customers,public.products

    # Snapshot behavior: what to do on first start
    # 'initial' = snapshot all existing rows, then stream changes
    # 'never'   = only stream changes from this point forward
    # 'always'  = re-snapshot on every connector restart (expensive!)
    snapshot.mode: initial

    # Topic naming: {server.name}.{schema}.{table}
    # e.g., production-db.public.orders
    topic.prefix: production-db

    # Heartbeat: keep WAL position advancing even when no changes occur
    # Prevents WAL accumulation on idle databases
    heartbeat.interval.ms: "10000"
    heartbeat.action.query: "UPDATE public.debezium_heartbeat SET last_heartbeat = NOW() WHERE id = 1"

    # Decimal handling: numbers → strings to avoid precision loss
    decimal.handling.mode: string

    # Timestamp handling: produce millisecond epoch timestamps
    time.precision.mode: adaptive_time_microseconds

    # Include transaction metadata in events
    provide.transaction.metadata: "true"

    # Offset storage: commit WAL position every 60 seconds or 10,000 events
    offset.flush.interval.ms: "60000"
    offset.flush.timeout.ms: "5000"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

curl -X POST http://kafka-connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d @postgres-connector.json

# Check connector status
curl http://kafka-connect:8083/connectors/postgres-orders-connector/status | jq .

# Expected output:
{
  "name": "postgres-orders-connector",
  "connector": {"state": "RUNNING", "worker_id": "..."},
  "tasks": [{"id": 0, "state": "RUNNING", "worker_id": "..."}],
  "type": "source"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# Use kafkacat/kcat to consume events
kcat -b kafka:9092 \
  -t production-db.public.orders \
  -C -o beginning \
  -f '%T: %s\n' | head -5 | jq .

# Expected output for an INSERT:
{
  "payload": {
    "before": null,
    "after": {
      "id": 1001,
      "customer_id": 42,
      "status": "pending",
      "amount": 4999,
      "created_at": "2026-03-27T10:00:00.000Z"
    },
    "op": "c",
    "source": {
      "table": "orders",
      "txId": 847234,
      "lsn": 24522983
    }
  }
}

1
2
3
4
5
6
7
8

-- MySQL: create replication user
CREATE USER 'debezium'@'%' IDENTIFIED WITH mysql_native_password BY 'secure-password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

-- Verify binlog format
SHOW VARIABLES LIKE 'binlog_format';  -- Must be ROW
SHOW VARIABLES LIKE 'binlog_row_image';  -- Must be FULL

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# mysql-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: mysql-inventory-connector
  namespace: kafka
  labels:
    strimzi.io/cluster: debezium-connect
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    database.hostname: mysql.databases.svc.cluster.local
    database.port: "3306"
    database.user: debezium
    database.password: ${file:/opt/kafka/external-configuration/debezium-secrets/mysql-password}
    database.server.id: "184054"     # Unique server ID (acts as a replica)
    topic.prefix: mysql-inventory

    # Capture specific databases and tables
    database.include.list: inventory
    table.include.list: inventory.products,inventory.stock_levels

    # Schema history: MySQL connector needs to track DDL changes
    # to correctly decode the binlog (schema at time of event may differ from now)
    schema.history.internal.kafka.topic: schema-changes.inventory
    schema.history.internal.kafka.bootstrap.servers: kafka:29092

    snapshot.mode: initial
    include.schema.changes: "true"   # Emit DDL changes as events too
    decimal.handling.mode: string

1
2
3

"transforms": "route",
"transforms.route.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.route.renames": "id:order_id,ts_ms:event_time"

1
2
3
4
5
6
7
8

config:
  # ExtractNewRecordState SMT: flatten the envelope to just the "after" value
  transforms: "unwrap"
  transforms.unwrap.type: "io.debezium.transforms.ExtractNewRecordState"
  transforms.unwrap.drop.tombstones: "false"    # Keep delete tombstones
  transforms.unwrap.delete.handling.mode: "rewrite"  # Add __deleted field
  transforms.unwrap.add.fields: "op,source.ts_ms,source.table"  # Keep metadata
  transforms.unwrap.add.headers: "op"           # Also add op to Kafka header

1

{"payload": {"before": {...}, "after": {"id": 1, "name": "Widget"}, "op": "u"}}

1

{"id": 1, "name": "Widget", "__op": "u", "__source_ts_ms": 1711584000123}

1
2
3
4
5

# Only forward events where status = 'shipped'
transforms: "filter"
transforms.filter.type: "io.debezium.transforms.Filter"
transforms.filter.language: "jsr223.groovy"
transforms.filter.condition: "value.after?.status == 'shipped'"

1
2
3
4
5
6
7

# Route order events to different topics based on status
transforms: "router"
transforms.router.type: "io.debezium.transforms.ContentBasedRouter"
transforms.router.language: "jsr223.groovy"
transforms.router.topic.expression: |
  'orders-' + (value.after?.status ?: 'unknown')
# Events land in: orders-pending, orders-shipped, orders-cancelled, etc.

1
2
3
4
5

# Replace PII fields with a hash before they reach Kafka
transforms: "maskPII"
transforms.maskPII.type: "org.apache.kafka.connect.transforms.MaskField$Value"
transforms.maskPII.fields: "credit_card_number,ssn,date_of_birth"
transforms.maskPII.replacement: "***REDACTED***"

1
2
3
4

transforms: "addTimestamp"
transforms.addTimestamp.type: "org.apache.kafka.connect.transforms.InsertHeader"
transforms.addTimestamp.header: "captured_at"
transforms.addTimestamp.value.literal: "${timestamp}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Use Avro with Schema Registry instead of JSON
# All consumers can read old and new events without code changes
config:
  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
  # BACKWARD compatibility: new schema can read old data
  # (safe for adding nullable fields)
  value.converter.schema.registry.auto.register.schemas: "true"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 1. Add new column with a default (backward compatible)
ALTER TABLE orders ADD COLUMN delivery_notes TEXT DEFAULT NULL;
-- Debezium automatically picks up the new field

-- 2. Never rename columns — add new, migrate, drop old
-- BAD:
-- ALTER TABLE orders RENAME COLUMN cust_id TO customer_id;  ← breaks consumers

-- GOOD (expand/contract):
ALTER TABLE orders ADD COLUMN customer_id BIGINT;        -- Step 1: add new
UPDATE orders SET customer_id = cust_id;                 -- Step 2: backfill
-- Update application to write both columns              -- Step 3: dual write
-- Update consumers to read new column                   -- Step 4: consumers migrate
ALTER TABLE orders DROP COLUMN cust_id;                  -- Step 5: drop old

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# Python consumer — detect and handle schema changes
from confluent_kafka import Consumer
import json

consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'order-processor',
    'auto.offset.reset': 'earliest',
})

consumer.subscribe(['production-db.public.orders'])

KNOWN_FIELDS = {'id', 'customer_id', 'status', 'amount', 'created_at'}

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue

    event = json.loads(msg.value())
    payload = event.get('payload', {})
    after = payload.get('after', {})

    if after is None:
        continue  # DELETE event

    # Detect new fields (schema evolution)
    new_fields = set(after.keys()) - KNOWN_FIELDS
    if new_fields:
        # Log but don't fail — forward-compatible consumers ignore unknown fields
        logger.info(f"New fields detected in orders: {new_fields}")

    # Process only known fields
    process_order({k: v for k, v in after.items() if k in KNOWN_FIELDS})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

# cache_invalidation_consumer.py
import json
import logging
from confluent_kafka import Consumer, KafkaError
import redis

log = logging.getLogger(__name__)

redis_client = redis.Redis(host='redis', decode_responses=True)

consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'cache-invalidator',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,    # Manual commit for at-least-once semantics
})

consumer.subscribe([
    'production-db.public.orders',
    'production-db.public.customers',
    'production-db.public.products',
])

def invalidate_cache(table: str, row_id: int, op: str):
    """Remove cached entries when the underlying row changes."""
    patterns = {
        'orders': [f"order:{row_id}", f"user_orders:{row_id}"],
        'customers': [f"customer:{row_id}", f"customer_profile:{row_id}"],
        'products': [f"product:{row_id}", f"product_list:*"],
    }
    keys = patterns.get(table, [])
    for key in keys:
        if '*' in key:
            # Scan for pattern matches (use sparingly)
            for k in redis_client.scan_iter(key):
                redis_client.delete(k)
        else:
            redis_client.delete(key)
    log.info(f"Invalidated cache for {table} id={row_id} op={op}")

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            log.error(f"Consumer error: {msg.error()}")
            continue

        event = json.loads(msg.value())
        payload = event.get('payload', {})
        op = payload.get('op')
        source = payload.get('source', {})
        table = source.get('table')

        # Get the ID from the after record (or before for deletes)
        record = payload.get('after') or payload.get('before')
        if record and table:
            invalidate_cache(table, record['id'], op)

        consumer.commit(msg)  # Commit after successful processing

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

// elasticsearch_sync.go
package main

import (
    "context"
    "encoding/json"
    "log/slog"

    "github.com/confluentinc/confluent-kafka-go/v2/kafka"
    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
)

type DebeziumPayload struct {
    Before map[string]any `json:"before"`
    After  map[string]any `json:"after"`
    Op     string         `json:"op"`
    Source struct {
        Table string `json:"table"`
    } `json:"source"`
}

func syncToElasticsearch(es *elasticsearch.Client, consumer *kafka.Consumer) {
    for {
        msg, err := consumer.ReadMessage(-1)
        if err != nil {
            slog.Error("read error", "err", err)
            continue
        }

        var event struct {
            Payload DebeziumPayload `json:"payload"`
        }
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            slog.Warn("unmarshal failed", "err", err)
            continue
        }

        p := event.Payload
        index := "orders" // Derived from topic or source table

        switch p.Op {
        case "c", "u", "r": // create, update, read (snapshot)
            id := fmt.Sprintf("%v", p.After["id"])
            body, _ := json.Marshal(p.After)

            req := esapi.IndexRequest{
                Index:      index,
                DocumentID: id,
                Body:       bytes.NewReader(body),
                Refresh:    "false", // Don't block on refresh
            }
            res, err := req.Do(context.Background(), es)
            if err != nil || res.IsError() {
                slog.Error("ES index failed", "id", id, "err", err)
            }

        case "d": // delete
            id := fmt.Sprintf("%v", p.Before["id"])
            req := esapi.DeleteRequest{
                Index:      index,
                DocumentID: id,
            }
            res, err := req.Do(context.Background(), es)
            if err != nil || res.IsError() {
                slog.Error("ES delete failed", "id", id, "err", err)
            }
        }

        consumer.CommitMessage(msg)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# Use the LSN (Log Sequence Number) as an idempotency key
# PostgreSQL LSN uniquely identifies each WAL record

processed_lsns = set()  # In practice: use Redis SET or DB table

def process_event(event: dict):
    lsn = event['payload']['source']['lsn']

    if lsn in processed_lsns:
        logger.info(f"Skipping duplicate event at LSN {lsn}")
        return

    # Process the event
    do_work(event)

    # Mark as processed
    processed_lsns.add(lsn)
    # Redis: redis_client.sadd("processed_lsns", lsn)
    # DB: INSERT INTO processed_events (lsn) ON CONFLICT DO NOTHING

1
2
3
4
5
6
7
8
9

# Connector lag: how far behind is Debezium from the latest WAL position?
# (exposed via Kafka Connect JMX → Prometheus via JMX Exporter)
kafka_connect_source_task_metrics_source_record_poll_rate

# Consumer group lag: how far behind are consumers from the latest event?
sum(kafka_consumergroup_lag) by (consumergroup, topic)

# Records captured per second
rate(kafka_connect_source_task_metrics_source_record_write_total[5m])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# Alert: CDC is falling behind (WAL accumulation risk)
- alert: DebeziumConsumerLagHigh
  expr: kafka_consumergroup_lag{consumergroup="debezium-connect"} > 100000
  for: 10m
  labels:
    severity: warning
  annotations:
    summary: "Debezium consumer lag {{ $value }} — CDC pipeline is behind"
    runbook_url: "https://runbooks.example.com/debezium/high-lag"

# Alert: Connector not running
- alert: DebeziumConnectorDown
  expr: kafka_connect_connector_status{connector="postgres-orders-connector"} != 1
  for: 2m
  labels:
    severity: critical
  annotations:
    summary: "Debezium connector {{ $labels.connector }} is not running"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- Monitor replication slot lag
SELECT
  slot_name,
  pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
  ) AS replication_lag,
  active
FROM pg_replication_slots;

-- If Debezium is down for extended period, consider dropping the slot
-- (you'll need to re-snapshot when it comes back)
-- SELECT pg_drop_replication_slot('debezium');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# PostgreSQL alert: replication slot WAL retention growing
- alert: PostgreSQLReplicationSlotLagHigh
  expr: |
    pg_replication_slots_pg_wal_lsn_diff_bytes{slot_name=~"debezium.*"} > 10737418240
  for: 15m
  labels:
    severity: critical
  annotations:
    summary: "Replication slot lag {{ $value | humanize1024 }}B — disk at risk"
    runbook_url: "https://runbooks.example.com/postgresql/replication-slot-lag"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Restart a failed connector
curl -X POST http://kafka-connect:8083/connectors/postgres-orders-connector/restart

# Restart a specific failed task
curl -X POST http://kafka-connect:8083/connectors/postgres-orders-connector/tasks/0/restart

# Check current offsets (WAL position Debezium has consumed to)
curl http://kafka-connect:8083/connectors/postgres-orders-connector/offsets

# Reset offsets to re-process from a specific point
# (useful after data loss or corruption)
curl -X DELETE http://kafka-connect:8083/connectors/postgres-orders-connector/offsets

1
2
3
4
5

config:
  # Enable exactly-once for the connector
  exactly.once.support: "required"
  # Requires Kafka 3.3+ and transactions enabled on the broker
  transaction.boundary: "poll"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

-- Add an outbox table to your schema
CREATE TABLE outbox (
    id          UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,  -- e.g., 'order'
    aggregate_id   VARCHAR(255) NOT NULL,  -- e.g., order ID
    event_type     VARCHAR(255) NOT NULL,  -- e.g., 'OrderShipped'
    payload        JSONB NOT NULL,
    created_at     TIMESTAMPTZ DEFAULT NOW()
);

-- Application: write to business table AND outbox in one transaction
BEGIN;
  UPDATE orders SET status = 'shipped' WHERE id = 12345;

  INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
  VALUES (
    'order',
    '12345',
    'OrderShipped',
    '{"order_id": 12345, "shipped_at": "2026-03-27T10:00:00Z", "tracking": "1Z999AA1"}'
  );
COMMIT;
-- If the transaction commits, BOTH the state change AND the event are durable.
-- Debezium reads the outbox table and publishes to Kafka.
-- No dual-write failure possible.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Debezium Outbox Event Router SMT
config:
  table.include.list: public.outbox
  transforms: "outbox"
  transforms.outbox.type: "io.debezium.transforms.outbox.EventRouter"
  transforms.outbox.table.field.event.id: "id"
  transforms.outbox.table.field.event.type: "event_type"
  transforms.outbox.table.field.event.key: "aggregate_id"
  transforms.outbox.table.field.event.payload: "payload"
  # Routes to topic named after aggregate_type: outbox.event.order
  transforms.outbox.route.by.field: "aggregate_type"
  transforms.outbox.route.topic.replacement: "outbox.event.$1"
  # Delete rows from outbox after capturing (keeps table small)
  transforms.outbox.table.tombstone.on.empty.payload: "true"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# Connector management (REST API)
curl http://connect:8083/connectors                                    # List all
curl http://connect:8083/connectors/<name>/status                     # Status
curl http://connect:8083/connectors/<name>/config                     # Config
curl -X POST http://connect:8083/connectors/<name>/restart            # Restart
curl -X DELETE http://connect:8083/connectors/<name>                  # Delete

# Validate connector config before creating
curl -X PUT http://connect:8083/connector-plugins/\
  io.debezium.connector.postgresql.PostgresConnector/config/validate \
  -H "Content-Type: application/json" -d @config.json

# Monitor consumer lag
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group my-consumer-group

# PostgreSQL: replication slot status
SELECT slot_name, active, restart_lsn,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag
FROM pg_replication_slots;

# Consume from a CDC topic
kcat -b kafka:9092 -t production-db.public.orders -C -o end | jq .