Change Data Capture com Debezium e Kafka

1. Visão geral

Quando um serviço precisa saber que os dados em outro banco de dados mudaram, a forma comum é escrever no banco de dados e depois enviar uma mensagem para o Kafka. Essa abordagem é chamada de dual-write e facilmente causa divergência de dados. Change Data Capture (CDC) resolve esse problema lendo diretamente o transaction log do banco de dados.

Neste artigo, você vai configurar PostgreSQL, Kafka e Debezium usando Docker Compose e registrar um conector. Depois você escreve um consumer Spring Boot para processar os eventos INSERT, UPDATE e DELETE da tabela orders.

2. Por que CDC é necessário

Veja o seguinte código de dual-write comum:

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

Este código tem dois pontos fracos:

• Se o Kafka falhar após save ser bem-sucedido, o banco de dados tem o pedido mas nenhum evento foi enviado.
• Se send for bem-sucedido mas a transação fizer rollback, o consumer receberá um evento de um pedido que não existe.

O CDC inverte esse fluxo. A aplicação apenas escreve no banco de dados. Um processo separado lê o transaction log (WAL com PostgreSQL, binlog com MySQL) e emite um evento para cada mudança que foi feita commit. Porque o log contém apenas mudanças que fizeram commit, o evento sempre corresponde aos dados reais.

Debezium é um conjunto de conectores CDC que rodam no Kafka Connect. O fluxo de dados fica assim:

Aplicação ──escreve──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Tópico Kafka ──> Consumer

3. Configurar o ambiente

3.1. Docker Compose

Você precisa de três containers: PostgreSQL com logical replication ativado, Kafka rodando no modo KRaft e Kafka Connect com o plugin Debezium pré-instalado.

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

Há três pontos a notar neste arquivo:

• wal_level=logical é obrigatório. Sem esse parâmetro, Debezium não consegue criar um replication slot.
• O Kafka tem dois listeners. O container usa kafka:29092, enquanto a aplicação Spring Boot rodando na sua máquina usa localhost:9092.
• O Kafka Connect salva a configuração, offset e status em três tópicos separados. Graças a isso, quando o conector é reiniciado, ele retoma a leitura do WAL do ponto anterior.

Inicie os containers e crie a tabela:

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 instrui o PostgreSQL a registrar a linha completa no WAL. Sem esse comando, o campo before em eventos UPDATE será null, e eventos DELETE conterão apenas a chave primária.

3.2. Registrar o conector

O Kafka Connect fornece uma API REST na porta 8083. Você registra o conector PostgreSQL assim:

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"
  }
}'

O significado das propriedades importantes:

• plugin.name: pgoutput usa o plugin de decodificação nativo do PostgreSQL 10+, então você não precisa instalar uma extensão adicional.
• topic.prefix junto com o esquema e nome da tabela criam o nome do tópico. Aqui o tópico é shop.public.orders.
• table.include.list restringe quais tabelas serão capturadas.
• decimal.handling.mode: string faz com que colunas NUMERIC sejam enviadas como strings "149.90". Por padrão, o Debezium codifica esse valor em bytes base64, que é muito difícil de ler.
• schemas.enable: false remove a seção schema que acompanha cada mensagem JSON, tornando o payload mais compacto.

Verifique o status do conector:

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. Estrutura do change event

Insira e depois atualize um pedido:

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;"

Depois leia o tópico usando o consumer console que vem na imagem do Kafka:

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

Um evento UPDATE fica assim (com o 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
}

O campo op indica o tipo de mudança:

op	Significado	before	after
c	INSERT	null	linha nova
u	UPDATE	linha antiga	linha nova
d	DELETE	linha antiga	null
r	Leitura no snapshot inicial	null	linha atual

Quando uma linha é deletada, o Debezium envia uma tombstone adicional, que é uma mensagem com a mesma chave mas valor null. A tombstone permite que o log compaction do Kafka limpe essa chave completamente. O seu consumer deve conseguir lidar com valor null.

5. Consumir eventos com Spring Boot

5.1. Dependências e configuração

<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>

A versão das dependências acima é gerenciada pelo 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

Aqui o consumer recebe o valor como String e faz o parsing manualmente. Essa abordagem oferece melhor controle sobre tombstone e erros de JSON em comparação com usar JsonDeserializer.

5.2. Modelo e 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 o campo source e qualquer novo campo que o Debezium possa adicionar no futuro. Jackson converte automaticamente a string "149.90" para BigDecimal, então você não perde precisão como faria com 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);
        }
    }
}

O parser retorna Optional.empty() para tombstone, então o listener pode simplesmente ignorar esse 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 o Spring passe null para o método quando encontra uma tombstone. Sem essa anotação, o Spring reportará um erro de conversão. O caso "r" é agrupado com "c" porque para o consumer, uma linha existente no snapshot inicial é como uma linha recém criada.

Executando a aplicação com os dados da seção 4, você verá:

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

6. Testando o parser

A lógica de parsing é a parte mais frágil quando o Debezium ou schema mudam. Você pode testar isso sem precisar do 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);
    }
}

O primeiro teste verifica o mapeamento do nome de coluna customer_email e do tipo BigDecimal. O segundo teste garante que tombstone não causa erro no listener. O terceiro teste confirma que JSON malformado lança uma exceção clara em vez de ser ignorado silenciosamente.

7. Erros comuns e quando não usar CDC

Aviso: O replication slot mantém o WAL até que o Debezium confirme que foi lido. Se o conector ficar parado por muitos dias, o WAL crescerá e pode preencher o disco do banco de dados. Você deve monitorar pg_replication_slots e deletar o slot quando remover o conector: SELECT pg_drop_replication_slot('shop_slot');.

Alguns outros erros que você encontrará:

• Consumer não é idempotente. Debezium e Kafka Connect garantem at-least-once. Após um restart, o mesmo evento pode chegar duas vezes. Use id junto com source.lsn para descartar duplicatas, ou implemente operações com semântica de upsert.
• Valores NUMERIC como base64 incompreensível. Você esqueceu de configurar decimal.handling.mode. Da mesma forma, colunas TIMESTAMP por padrão são enviadas como números de microsegundos desde a época, então você precisa converter manualmente.
• before sempre null. A tabela não tem REPLICA IDENTITY FULL definido.
• Snapshot inicial demora muito. Com tabelas grandes, snapshot.mode=initial lê a tabela inteira antes de começar a fazer streaming. Considere no_data se você só precisa das mudanças a partir de agora, ou use incremental snapshot.
• Vazamento de dados internos. O evento reflete a estrutura exata da tabela. Se você enviar diretamente para outro serviço, está expondo o schema do seu banco de dados. Para eventos de negócio, combine Debezium com o Outbox pattern: a aplicação escreve em uma tabela outbox na mesma transação, e o Debezium só captura essa tabela.
• Usar a conta postgres em produção. Crie um usuário dedicado com permissões REPLICATION e SELECT apenas nas tabelas que precisa capturar.

Você não deve usar CDC quando:

• Tem apenas uma aplicação e ela mesma consegue gerar eventos no mesmo processo. Nesse caso, um transactional outbox com um simples poller é suficiente.
• Você precisa de eventos com significado de negócio como “OrderPaid” em vez de “linha X mudou coluna status”. CDC é apenas sobre mudanças de dados, não sobre a razão.
• Seu time de operações ainda não está pronto para gerenciar Kafka Connect, replication slots e evolução de schema.

8. Conclusão

Neste artigo, você configurou PostgreSQL com logical replication, Kafka e Debezium, depois registrou um conector para capturar a tabela orders. Você também aprendeu sobre a estrutura do change event com os campos before, after e op, e então escreveu um consumer Spring Boot que processa tombstone junto com testes para a parte de parsing.

CDC elimina o risco de dual-write porque o banco de dados se torna a única fonte da verdade. Em troca, você precisa monitorar o replication slot, escrever consumers idempotentes e considerar usar o Outbox pattern quando precisa de eventos de negócio bem definidos.