Change Data Capture with Debezium — Real-Time Data Sync for Microservices

{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 1001,
      "name": "Nguyen Van A",
      "email": "a@example.com",
      "status": "active"
    },
    "after": {
      "id": 1001,
      "name": "Nguyen Van A",
      "email": "newemail@example.com",
      "status": "active"
    },
    "source": {
      "version": "2.7.0",
      "connector": "postgresql",
      "name": "myapp",
      "ts_ms": 1713436800000,
      "db": "mydb",
      "schema": "public",
      "table": "users",
      "lsn": 123456789,
      "txId": 5678
    },
    "op": "u",
    "ts_ms": 1713436800123,
    "transaction": {
      "id": "5678",
      "total_order": 3,
      "data_collection_order": 1
    }
  }
}

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: appuser
      POSTGRES_PASSWORD: secret
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"
      - "-c"
      - "max_replication_slots=4"
      - "-c"
      - "max_wal_senders=4"
    ports:
      - "5432:5432"

  kafka:
    image: bitnami/kafka:3.7
    environment:
      KAFKA_CFG_NODE_ID: 0
      KAFKA_CFG_PROCESS_ROLES: controller,broker
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
    ports:
      - "9092:9092"

  connect:
    image: debezium/connect:2.7
    depends_on:
      - kafka
      - postgres
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: cdc-connect-cluster
      CONFIG_STORAGE_TOPIC: _connect-configs
      OFFSET_STORAGE_TOPIC: _connect-offsets
      STATUS_STORAGE_TOPIC: _connect-status
    ports:
      - "8083:8083"

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "myapp-postgres-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "appuser",
      "database.password": "secret",
      "database.dbname": "myapp",
      "topic.prefix": "myapp",
      "schema.include.list": "public",
      "table.include.list": "public.users,public.orders",
      "plugin.name": "pgoutput",
      "slot.name": "debezium_slot",
      "publication.name": "debezium_pub",
      "snapshot.mode": "initial",
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
      "transforms.unwrap.drop.tombstones": "false",
      "transforms.unwrap.delete.handling.mode": "rewrite",
      "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"
    }
  }'

-- Outbox table schema
CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Example: creating an order
BEGIN;
  INSERT INTO orders (id, customer_id, total, status)
    VALUES ('ord-001', 'cust-123', 599000, 'pending');

  INSERT INTO outbox (aggregate_type, aggregate_id, type, payload)
    VALUES (
      'Order', 'ord-001', 'OrderCreated',
      '{"orderId":"ord-001","customerId":"cust-123","total":599000}'::jsonb
    );
COMMIT;

{
  "transforms": "outbox",
  "transforms.outbox.type":
    "io.debezium.transforms.outbox.EventRouter",
  "transforms.outbox.table.field.event.id": "id",
  "transforms.outbox.table.field.event.key": "aggregate_id",
  "transforms.outbox.table.field.event.type": "type",
  "transforms.outbox.table.field.event.payload": "payload",
  "transforms.outbox.route.by.field": "aggregate_type",
  "transforms.outbox.route.topic.regex": "(.*)Order(.*)",
  "transforms.outbox.route.topic.replacement": "order-events"
}

// .NET consumer that invalidates Redis cache from CDC events
public class CdcCacheInvalidator : BackgroundService
{
    private readonly IConsumer<string, string> _consumer;
    private readonly IConnectionMultiplexer _redis;

    protected override async Task ExecuteAsync(
        CancellationToken stoppingToken)
    {
        _consumer.Subscribe("myapp.public.products");

        while (!stoppingToken.IsCancellationRequested)
        {
            var result = _consumer.Consume(stoppingToken);
            var change = JsonSerializer
                .Deserialize<DebeziumEvent>(result.Message.Value);

            var db = _redis.GetDatabase();
            var productId = change.After?.Id ?? change.Before?.Id;

            switch (change.Op)
            {
                case "c":
                case "u":
                    // Upsert into cache with TTL
                    await db.StringSetAsync(
                        $"product:{productId}",
                        JsonSerializer.Serialize(change.After),
                        TimeSpan.FromMinutes(30));
                    break;

                case "d":
                    // Remove from cache
                    await db.KeyDeleteAsync($"product:{productId}");
                    break;
            }
        }
    }
}

# application.properties for Debezium Server
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.database.hostname=localhost
debezium.source.database.port=5432
debezium.source.database.user=appuser
debezium.source.database.password=secret
debezium.source.database.dbname=myapp
debezium.source.topic.prefix=myapp
debezium.source.plugin.name=pgoutput
debezium.source.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
debezium.source.offset.storage.file.filename=/data/offsets.dat

# Sink: Redis Streams
debezium.sink.type=redis
debezium.sink.redis.address=localhost:6379

# Or sink: HTTP Webhook
# debezium.sink.type=http
# debezium.sink.http.url=https://api.example.com/webhooks/cdc

# Prometheus scrape config for Debezium metrics
- job_name: 'debezium'
  metrics_path: '/metrics'
  static_configs:
    - targets: ['connect:8083']

# Important alert rules
groups:
  - name: debezium_alerts
    rules:
      - alert: DebeziumHighLag
        expr: debezium_metrics_MilliSecondsBehindSource > 30000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Debezium lag > 30s on {{ $labels.context }}"

      - alert: DebeziumConnectorDown
        expr: debezium_metrics_Connected == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Debezium connector lost DB connection"

// Program.cs - Register the CDC consumer service
builder.Services.AddSingleton<IConsumer<string, string>>(sp =>
{
    var config = new ConsumerConfig
    {
        BootstrapServers = "kafka:9092",
        GroupId = "order-projection-service",
        AutoOffsetReset = AutoOffsetReset.Earliest,
        EnableAutoCommit = false, // Manual commit for exactly-once
        EnablePartitionEof = true
    };
    return new ConsumerBuilder<string, string>(config).Build();
});

builder.Services.AddHostedService<OrderProjectionService>();

// OrderProjectionService.cs
public class OrderProjectionService : BackgroundService
{
    private readonly IConsumer<string, string> _consumer;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<OrderProjectionService> _logger;

    protected override async Task ExecuteAsync(
        CancellationToken ct)
    {
        _consumer.Subscribe(new[] {
            "myapp.public.orders",
            "myapp.public.order_items"
        });

        while (!ct.IsCancellationRequested)
        {
            try
            {
                var result = _consumer.Consume(ct);
                if (result.IsPartitionEOF) continue;

                var evt = JsonSerializer
                    .Deserialize<CdcEvent>(result.Message.Value);

                using var scope = _scopeFactory.CreateScope();
                var handler = scope.ServiceProvider
                    .GetRequiredService<ICdcEventHandler>();

                await handler.HandleAsync(evt);

                // Commit offset after successful processing
                _consumer.Commit(result);
            }
            catch (ConsumeException ex)
            {
                _logger.LogError(ex,
                    "CDC consume error on {Topic}",
                    ex.ConsumerRecord?.Topic);
                // Implement retry / dead letter logic
            }
        }
    }
}

// ICdcEventHandler implementation
public class OrderCdcHandler : ICdcEventHandler
{
    private readonly ElasticClient _elastic;

    public async Task HandleAsync(CdcEvent evt)
    {
        if (evt.Source.Table == "orders")
        {
            switch (evt.Op)
            {
                case "c":
                case "u":
                    await _elastic.IndexDocumentAsync(
                        MapToSearchModel(evt.After));
                    break;
                case "d":
                    await _elastic.DeleteAsync<OrderSearch>(
                        evt.Before.Id);
                    break;
            }
        }
    }
}