Change Data Capture con Debezium y Kafka

1. Descripción general

Cuando un servicio necesita saber que los datos en otra base de datos acaban de cambiar, la forma común es escribir en la base de datos y luego enviar un mensaje adicional a Kafka. Esto se llama dual-write y es propenso a inconsistencias de datos. Change Data Capture (CDC) resuelve este problema leyendo directamente el transaction log de la base de datos.

En este artículo, configurarás PostgreSQL, Kafka y Debezium con Docker Compose, luego registrarás un connector. Después escribirás un consumer de Spring Boot para procesar eventos INSERT, UPDATE y DELETE de la tabla orders.

2. Por qué necesitas CDC

Observa el siguiente fragmento de código dual-write que ves con frecuencia:

@Transactional
public void placeOrder(Order order) {
    orderRepository.save(order);
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
}

Este código tiene dos puntos débiles:

• Si Kafka falla después de que save tenga éxito, la base de datos tiene la orden pero ningún evento se envía.
• Si send tiene éxito pero la transacción hace rollback, el consumer recibe un evento de una orden que no existe.

CDC invierte este flujo. La aplicación solo escribe en la base de datos. Un proceso separado lee el transaction log (WAL en PostgreSQL, binlog en MySQL) y emite un evento por cada cambio confirmado. Como el log solo contiene cambios confirmados, el evento siempre coincide con los datos reales.

Debezium es un conjunto de connectors de CDC que se ejecutan en Kafka Connect. El flujo de datos es así:

Aplicación ──escribe──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> topic de Kafka ──> Consumer

3. Configurar el entorno

3.1. Docker Compose

Necesitas tres contenedores: PostgreSQL con replicación lógica habilitada, Kafka ejecutándose en modo KRaft y Kafka Connect con el plugin de Debezium incluido.

services:
  postgres:
    image: postgres:16
    command: ["postgres", "-c", "wal_level=logical"]
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    ports:
      - "5432:5432"

  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      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: quay.io/debezium/connect:2.7
    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_statuses

Hay tres cosas a tener en cuenta en este archivo:

• wal_level=logical es obligatorio. Sin este parámetro, Debezium no puede crear un replication slot.
• Kafka tiene dos listeners. El contenedor usa kafka:29092, mientras que la aplicación Spring Boot que corre en tu máquina usa localhost:9092.
• Kafka Connect guarda la configuración, offset y estado en tres topics separados. Gracias a esto, el connector después de reiniciarse seguirá leyendo desde la posición anterior en el WAL.

Inicia y crea la tabla:

docker compose up -d
docker compose exec postgres psql -U postgres -d shop -c "
  CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer_email VARCHAR(255) NOT NULL,
    status VARCHAR(32) NOT NULL,
    total NUMERIC(10,2) NOT NULL
  );
  ALTER TABLE orders REPLICA IDENTITY FULL;"

REPLICA IDENTITY FULL le dice a PostgreSQL que escriba la fila completa en el WAL. Sin este comando, el campo before en el evento UPDATE será null, y el evento DELETE solo contendrá la clave primaria.

3.2. Registrar el connector

Kafka Connect proporciona una API REST en el puerto 8083. Registras el connector de PostgreSQL así:

curl -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "shop_slot",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "string",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}'

El significado de las propiedades importantes:

• plugin.name: pgoutput usa el plugin de decodificación incorporado en PostgreSQL 10+, por lo que no necesitas instalar una extensión adicional.
• topic.prefix junto con el esquema y el nombre de la tabla crean el nombre del topic. Aquí el topic es shop.public.orders.
• table.include.list limita qué tablas se capturan.
• decimal.handling.mode: string hace que las columnas NUMERIC se envíen como cadenas "149.90". Por defecto, Debezium codifica este valor como bytes en base64, que es difícil de leer.
• schemas.enable: false elimina la parte schema que acompaña cada mensaje JSON, haciendo el payload más compacto.

Verifica el estado del connector:

curl -s localhost:8083/connectors/shop-connector/status

{"name":"shop-connector","connector":{"state":"RUNNING","worker_id":"172.18.0.4:8083"},
 "tasks":[{"id":0,"state":"RUNNING","worker_id":"172.18.0.4:8083"}],"type":"source"}

4. Estructura del change event

Inserta y luego actualiza una orden:

docker compose exec postgres psql -U postgres -d shop -c "
  INSERT INTO orders (customer_email, status, total) VALUES ('an@example.com', 'NEW', 149.90);
  UPDATE orders SET status = 'PAID' WHERE id = 1;"

Luego lee el topic usando el console consumer incluido en la imagen de Kafka:

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic shop.public.orders --from-beginning

El evento UPDATE se verá así (con el campo source abreviado):

{
  "before": {"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
  "after":  {"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
  "source": {"connector":"postgresql","db":"shop","table":"orders","lsn":24023128, ...},
  "op": "u",
  "ts_ms": 1760170000123
}

El campo op indica el tipo de cambio:

op	Significado	before	after
c	INSERT	null	fila nueva
u	UPDATE	fila antigua	fila nueva
d	DELETE	fila antigua	null
r	Lectura en el snapshot inicial	null	fila existente

Cuando eliminas una fila, Debezium envía un tombstone adicional, es decir, un mensaje con la misma clave pero con value null. El tombstone permite que la log compaction de Kafka limpie completamente esa clave. Tu consumer debe ser capaz de manejar un value null.

5. Consumir eventos con Spring Boot

5.1. Dependencias y configuración

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Las versiones de estas dependencias son gestionadas por spring-boot-starter-parent 3.x.

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-sync
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

Aquí el consumer recibe el value como String y lo analiza por sí mismo. Este enfoque te da más control sobre los tombstones y errores de JSON en comparación con usar JsonDeserializer.

5.2. Modelo y parser

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderRow(
        Long id,
        @JsonProperty("customer_email") String customerEmail,
        String status,
        BigDecimal total) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
        OrderRow before,
        OrderRow after,
        String op,
        @JsonProperty("ts_ms") Long tsMs) {
}

@JsonIgnoreProperties(ignoreUnknown = true) ignora el campo source y cualquier otro campo nuevo que Debezium pueda agregar en el futuro. Jackson convierte automáticamente la cadena "149.90" a BigDecimal, así que no pierdes precisión como lo harías con double.

@Component
public class OrderChangeParser {

    private final ObjectMapper objectMapper;

    public OrderChangeParser(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public Optional<OrderChangeEvent> parse(String value) {
        if (value == null) {
            return Optional.empty(); // tombstone
        }
        try {
            return Optional.of(objectMapper.readValue(value, OrderChangeEvent.class));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Invalid Debezium event", e);
        }
    }
}

El parser devuelve Optional.empty() para tombstones, así que el listener solo necesita omitir este caso.

5.3. Listener

@Component
public class OrderCdcListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCdcListener.class);
    private final OrderChangeParser parser;

    public OrderCdcListener(OrderChangeParser parser) {
        this.parser = parser;
    }

    @KafkaListener(topics = "shop.public.orders")
    public void onOrderChange(@Payload(required = false) String value) {
        parser.parse(value).ifPresent(event -> {
            switch (event.op()) {
                case "c", "r" -> log.info("Order created: {}", event.after());
                case "u" -> log.info("Order {} status {} -> {}",
                        event.after().id(), event.before().status(), event.after().status());
                case "d" -> log.info("Order deleted: {}", event.before().id());
                default -> log.warn("Unknown op: {}", event.op());
            }
        });
    }
}

@Payload(required = false) permite que Spring pase null al método cuando encuentra un tombstone. Sin esta anotación, Spring reportará un error de conversión. El caso "r" se agrupa con "c" porque para un consumer, una fila existente en el snapshot es lo mismo que una fila que acaba de crearse.

Ejecuta la aplicación con los datos de la sección 4 y verás:

INFO  OrderCdcListener : Order created: OrderRow[id=1, customerEmail=an@example.com, status=NEW, total=149.90]
INFO  OrderCdcListener : Order 1 status NEW -> PAID

6. Probar el parser

La lógica de análisis es la parte más propensa a romperse cuando Debezium o el esquema cambian. Puedes probar esta parte sin necesidad de Kafka:

class OrderChangeParserTest {

    private final OrderChangeParser parser = new OrderChangeParser(new ObjectMapper());

    @Test
    void parsesUpdateEvent() {
        String json = """
            {"before":{"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
             "after":{"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
             "source":{"connector":"postgresql"},
             "op":"u","ts_ms":1760170000123}
            """;

        OrderChangeEvent event = parser.parse(json).orElseThrow();

        assertThat(event.op()).isEqualTo("u");
        assertThat(event.before().status()).isEqualTo("NEW");
        assertThat(event.after().status()).isEqualTo("PAID");
        assertThat(event.after().total()).isEqualByComparingTo("149.90");
    }

    @Test
    void returnsEmptyForTombstone() {
        assertThat(parser.parse(null)).isEmpty();
    }

    @Test
    void rejectsMalformedJson() {
        assertThatThrownBy(() -> parser.parse("{not json"))
                .isInstanceOf(IllegalArgumentException.class);
    }
}

La primera prueba verifica la asignación del nombre de columna customer_email y el tipo BigDecimal. La segunda prueba asegura que los tombstones no causen que el listener falle. La tercera prueba confirma que el JSON malformado se reporta claramente en lugar de ser silenciado.

7. Errores comunes y cuándo no usar CDC

Advertencia: El replication slot retiene el WAL hasta que Debezium confirma que lo ha leído. Si el connector se detiene durante varios días, el WAL crecerá y puede llenar el disco de la base de datos. Necesitas monitorear pg_replication_slots y eliminar el slot cuando desinstales el connector: SELECT pg_drop_replication_slot('shop_slot');.

Algunos otros errores que encontrarás con frecuencia:

• Consumer no es idempotente. Debezium y Kafka Connect garantizan at-least-once. Después de un reinicio, el mismo evento puede llegar dos veces. Usa id junto con source.lsn para descartar duplicados, o diseña la operación como un upsert.
• El valor NUMERIC se convierte a una cadena base64 incomprensible. Olvidaste configurar decimal.handling.mode. De manera similar, las columnas TIMESTAMP por defecto se envían como un número de microsegundos desde epoch, así que necesitas convertirlo tú mismo.
• before siempre es null. La tabla no tiene REPLICA IDENTITY FULL configurado.
• El snapshot inicial es muy lento. Con tablas grandes, snapshot.mode=initial leerá toda la tabla antes de hacer streaming. Considera no_data si solo necesitas cambios a partir de ahora, o usa incremental snapshot.
• Fuga de datos internos. El evento refleja la estructura exacta de la tabla. Si lo pasas directamente a otro servicio, estás exponiendo el esquema de la base de datos. Para eventos de negocio, combina Debezium con el patrón Outbox: la aplicación escribe en una tabla outbox en la misma transacción, y Debezium solo captura esa tabla.
• Usar la cuenta postgres en producción. Crea un usuario separado con permisos REPLICATION y SELECT solo en las tablas que necesitas capturar.

No debes usar CDC cuando:

• Solo tienes una aplicación y puede generar eventos en el mismo proceso. En este caso, un outbox transaccional con un poller simple es suficiente.
• Necesitas eventos con significado de negocio como “OrderPaid” en lugar de “la fila X cambió la columna status”. CDC solo sabe que los datos cambiaron, no por qué.
• Tu equipo de operaciones aún no está listo para gestionar Kafka Connect, replication slots y la evolución del esquema.

8. Conclusión

En este artículo, configuraste PostgreSQL con replicación lógica, Kafka y Debezium, luego registraste un connector para capturar la tabla orders. También aprendiste la estructura del change event con los campos before, after y op, y luego escribiste un consumer de Spring Boot que maneja tanto los tombstones como las pruebas para la parte de análisis.

CDC elimina el riesgo del dual-write porque la base de datos se convierte en la única fuente de verdad. A cambio, debes monitorear el replication slot, escribir consumers idempotentes y considerar usar el patrón Outbox cuando necesites eventos de negocio claros.