Microservices Data Synchronization with the Outbox Pattern and CDC: A Practical Debezium Guide

// Dangerous Dual Write pattern
@Transactional
public Order createOrder(OrderRequest request) {
    Order order = orderRepository.save(new Order(request));  // Step 1: Save to DB

    // Step 2: Publish event to message broker
    kafkaTemplate.send("order-events", new OrderCreatedEvent(order));

    return order;
}

[Service Code]
    │
    ├── Business data INSERT ────────┐
    │                                │  Same DB transaction
    └── Outbox table INSERT ─────────┘
                                      │
                                      ▼
                              [Outbox Relay]
                              (CDC or Polling)
                                      │
                                      ▼
                              [Message Broker]
                              (Kafka, RabbitMQ)
                                      │
                                      ▼
                              [Other Microservices]

-- PostgreSQL Outbox table DDL
CREATE TABLE outbox_events (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type  VARCHAR(255) NOT NULL,   -- Domain the event belongs to (e.g., 'Order')
    aggregate_id    VARCHAR(255) NOT NULL,   -- Domain entity ID (e.g., order ID)
    event_type      VARCHAR(255) NOT NULL,   -- Event type (e.g., 'OrderCreated')
    payload         JSONB NOT NULL,          -- Event body
    metadata        JSONB DEFAULT '{}',      -- Additional metadata
    created_at      TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMP WITH TIME ZONE,-- For polling: timestamp when published
    retry_count     INT DEFAULT 0,           -- For polling: retry count
    status          VARCHAR(20) DEFAULT 'PENDING'  -- PENDING, PUBLISHED, FAILED
);

-- Index for querying unpublished events in polling approach
CREATE INDEX idx_outbox_status_created ON outbox_events(status, created_at)
    WHERE status = 'PENDING';

-- For CDC approach, the status/published_at columns are unnecessary,
-- and rows can be deleted after event publication

-- Automatic Outbox recording using triggers in MySQL
DELIMITER //

CREATE TRIGGER after_order_insert
AFTER INSERT ON orders
FOR EACH ROW
BEGIN
    INSERT INTO outbox_events (
        id, aggregate_type, aggregate_id, event_type, payload, created_at
    ) VALUES (
        UUID(),
        'Order',
        NEW.order_id,
        'OrderCreated',
        JSON_OBJECT(
            'orderId', NEW.order_id,
            'userId', NEW.user_id,
            'totalAmount', NEW.total_amount,
            'status', NEW.status,
            'createdAt', NEW.created_at
        ),
        NOW()
    );
END //

CREATE TRIGGER after_order_update
AFTER UPDATE ON orders
FOR EACH ROW
BEGIN
    IF OLD.status != NEW.status THEN
        INSERT INTO outbox_events (
            id, aggregate_type, aggregate_id, event_type, payload, created_at
        ) VALUES (
            UUID(),
            'Order',
            NEW.order_id,
            CONCAT('OrderStatus', NEW.status),
            JSON_OBJECT(
                'orderId', NEW.order_id,
                'previousStatus', OLD.status,
                'newStatus', NEW.status,
                'updatedAt', NEW.updated_at
            ),
            NOW()
        );
    END IF;
END //

DELIMITER ;

# docker-compose.yml - Full Debezium + Kafka stack
version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: orderdb
      POSTGRES_USER: appuser
      POSTGRES_PASSWORD: secret
    command:
      - 'postgres'
      - '-c'
      - 'wal_level=logical' # Required setting for CDC
      - '-c'
      - 'max_replication_slots=4'
      - '-c'
      - 'max_wal_senders=4'
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  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_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: debezium/connect:2.7
    depends_on:
      - kafka
      - postgres
    ports:
      - '8083:8083'
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: debezium-connect-group
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1

  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://connect:8083

volumes:
  postgres_data:

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "appuser",
    "database.password": "secret",
    "database.dbname": "orderdb",
    "topic.prefix": "order-service",
    "schema.include.list": "public",
    "table.include.list": "public.outbox_events",

    "tombstones.on.delete": "false",
    "slot.name": "order_outbox_slot",
    "plugin.name": "pgoutput",

    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.table.field.event.timestamp": "created_at",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "events.${routedByValue}",

    "transforms.outbox.table.expand.json.payload": "true",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    "heartbeat.interval.ms": "10000",
    "snapshot.mode": "initial"
  }
}

# Register connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @order-outbox-connector.json

# Check connector status
curl -s http://localhost:8083/connectors/order-outbox-connector/status | jq .

# Restart connector
curl -X POST http://localhost:8083/connectors/order-outbox-connector/restart

# List topics (via Kafka UI or CLI)
docker exec -it kafka kafka-topics --bootstrap-server localhost:9092 --list

// OutboxEvent.java - Outbox entity
@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(name = "aggregate_type", nullable = false)
    private String aggregateType;

    @Column(name = "aggregate_id", nullable = false)
    private String aggregateId;

    @Column(name = "event_type", nullable = false)
    private String eventType;

    @Column(name = "payload", columnDefinition = "jsonb", nullable = false)
    private String payload;

    @Column(name = "created_at", nullable = false)
    private Instant createdAt;

    // Default constructor, getters, setters omitted

    public static OutboxEvent create(String aggregateType, String aggregateId,
                                     String eventType, Object payload) {
        OutboxEvent event = new OutboxEvent();
        event.aggregateType = aggregateType;
        event.aggregateId = aggregateId;
        event.eventType = eventType;
        event.payload = toJson(payload);
        event.createdAt = Instant.now();
        return event;
    }

    private static String toJson(Object obj) {
        try {
            return new ObjectMapper()
                .registerModule(new JavaTimeModule())
                .writeValueAsString(obj);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("JSON serialization failed", e);
        }
    }
}

// OrderService.java - Order service with Outbox Pattern applied
@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;

    @Transactional  // Single transaction persists both business data and event
    public Order createOrder(OrderRequest request) {
        // 1. Execute business logic
        Order order = Order.builder()
            .userId(request.getUserId())
            .items(request.getItems())
            .totalAmount(calculateTotal(request.getItems()))
            .status(OrderStatus.CREATED)
            .build();

        Order savedOrder = orderRepository.save(order);

        // 2. Record event in Outbox table (same transaction)
        OrderCreatedPayload eventPayload = OrderCreatedPayload.builder()
            .orderId(savedOrder.getId().toString())
            .userId(savedOrder.getUserId())
            .totalAmount(savedOrder.getTotalAmount())
            .items(savedOrder.getItems())
            .createdAt(savedOrder.getCreatedAt())
            .build();

        OutboxEvent outboxEvent = OutboxEvent.create(
            "Order",                              // aggregate_type -> determines Kafka topic
            savedOrder.getId().toString(),         // aggregate_id -> Kafka partition key
            "OrderCreated",                        // event_type
            eventPayload                           // payload
        );

        outboxRepository.save(outboxEvent);

        return savedOrder;
    }

    @Transactional
    public Order cancelOrder(UUID orderId) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));

        if (order.getStatus() != OrderStatus.CREATED) {
            throw new IllegalStateException(
                "Only orders in CREATED status can be cancelled. Current status: " + order.getStatus()
            );
        }

        order.setStatus(OrderStatus.CANCELLED);
        Order savedOrder = orderRepository.save(order);

        OutboxEvent outboxEvent = OutboxEvent.create(
            "Order",
            savedOrder.getId().toString(),
            "OrderCancelled",
            Map.of(
                "orderId", savedOrder.getId().toString(),
                "reason", "User request",
                "cancelledAt", Instant.now().toString()
            )
        );

        outboxRepository.save(outboxEvent);

        return savedOrder;
    }
}

import json
import uuid
from datetime import datetime
from kafka import KafkaConsumer
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

# Processing history table for idempotency guarantees
# CREATE TABLE processed_events (
#     event_id UUID PRIMARY KEY,
#     processed_at TIMESTAMP NOT NULL DEFAULT NOW()
# );

engine = create_engine('postgresql://user:pass@localhost:5432/inventorydb')

consumer = KafkaConsumer(
    'events.Order',
    bootstrap_servers=['localhost:9092'],
    group_id='inventory-service',
    auto_offset_reset='earliest',
    enable_auto_commit=False,          # Manual commit for precise control
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    max_poll_records=100,
)

def process_order_event(event_data: dict, event_id: str) -> None:
    """Process events with idempotency guarantees"""
    with Session(engine) as session:
        with session.begin():
            # 1. Check if the event has already been processed
            result = session.execute(
                text("SELECT 1 FROM processed_events WHERE event_id = :id"),
                {"id": event_id}
            )
            if result.fetchone():
                print(f"Skipping already processed event: {event_id}")
                return

            # 2. Execute business logic
            event_type = event_data.get('eventType', '')

            if event_type == 'OrderCreated':
                items = event_data.get('items', [])
                for item in items:
                    session.execute(
                        text("""
                            UPDATE inventory
                            SET reserved_quantity = reserved_quantity + :qty
                            WHERE product_id = :product_id
                              AND available_quantity >= :qty
                        """),
                        {"qty": item['quantity'], "product_id": item['productId']}
                    )

            elif event_type == 'OrderCancelled':
                order_id = event_data.get('orderId')
                session.execute(
                    text("""
                        UPDATE inventory i
                        SET reserved_quantity = reserved_quantity - oi.quantity
                        FROM order_items oi
                        WHERE oi.order_id = :order_id
                          AND oi.product_id = i.product_id
                    """),
                    {"order_id": order_id}
                )

            # 3. Record processing history (same transaction)
            session.execute(
                text("""
                    INSERT INTO processed_events (event_id, processed_at)
                    VALUES (:id, :now)
                """),
                {"id": event_id, "now": datetime.utcnow()}
            )

    print(f"Event processed successfully: {event_id} ({event_type})")


# Main consumer loop
print("Starting event consumer...")
try:
    for message in consumer:
        try:
            event_data = message.value
            # Extract event_id from headers set by Debezium Outbox Event Router
            headers = {k: v.decode('utf-8') for k, v in message.headers}
            event_id = headers.get('id', str(uuid.uuid4()))

            process_order_event(event_data, event_id)

            # Commit offset only after successful processing
            consumer.commit()

        except Exception as e:
            print(f"Event processing failed: {e}")
            # On failure, send to DLQ (Dead Letter Queue) or apply retry logic
            # Here we simply log and move on to the next event

except KeyboardInterrupt:
    print("Consumer shutting down")
finally:
    consumer.close()

# 1. Check connector status
curl -s http://localhost:8083/connectors/order-outbox-connector/status | jq .

# 2. Attempt to restart the connector task
curl -X POST http://localhost:8083/connectors/order-outbox-connector/tasks/0/restart

# 3. If restart fails, delete and re-register the connector
curl -X DELETE http://localhost:8083/connectors/order-outbox-connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @order-outbox-connector.json

# 4. Check PostgreSQL WAL slot status
psql -c "SELECT * FROM pg_replication_slots;"

# 5. Re-create slot if necessary
psql -c "SELECT pg_drop_replication_slot('order_outbox_slot');"
# The slot is automatically created when the connector is re-registered

-- Daily partitioned Outbox table
CREATE TABLE outbox_events (
    id              UUID NOT NULL DEFAULT gen_random_uuid(),
    aggregate_type  VARCHAR(255) NOT NULL,
    aggregate_id    VARCHAR(255) NOT NULL,
    event_type      VARCHAR(255) NOT NULL,
    payload         JSONB NOT NULL,
    created_at      TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- Automatic daily partition creation (using pg_partman)
SELECT partman.create_parent(
    'public.outbox_events',
    'created_at',
    'native',
    'daily'
);

-- Automatically drop partitions older than 7 days
UPDATE partman.part_config
SET retention = '7 days', retention_keep_table = false
WHERE parent_table = 'public.outbox_events';