VI
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish日本語한국어简体中文
Trang chủ / Kiến trúc & Thiết kế / Kafka

Change Data Capture với Debezium và Kafka

Cris The Dev · 12/10/2026 · 8 phút đọc · 312

1. Tổng quan

Khi một service cần biết dữ liệu ở database khác vừa thay đổi, cách thường thấy là ghi vào database rồi gửi thêm một message lên Kafka. Cách này gọi là dual-write và dễ gây lệch dữ liệu. Change Data Capture (CDC) giải quyết vấn đề này bằng cách đọc thẳng transaction log của database.

Trong bài này, bạn sẽ dựng PostgreSQL, Kafka và Debezium bằng Docker Compose, rồi đăng ký một connector. Sau đó bạn viết một consumer Spring Boot để xử lý các sự kiện INSERT, UPDATE và DELETE của bảng orders.

2. Vì sao cần CDC

Hãy xem đoạn code dual-write quen thuộc sau:

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

Đoạn code này có hai điểm yếu:

  • Nếu Kafka lỗi sau khi save thành công, database có đơn hàng nhưng không có event nào được gửi đi.
  • Nếu send thành công nhưng transaction rollback, consumer nhận được event của một đơn hàng không tồn tại.

CDC đảo ngược luồng này. Ứng dụng chỉ ghi vào database. Một tiến trình riêng đọc transaction log (WAL với PostgreSQL, binlog với MySQL) và phát event cho mỗi thay đổi đã commit. Vì log chỉ chứa thay đổi đã commit, event luôn khớp với dữ liệu thật.

Debezium là một bộ connector CDC chạy trên Kafka Connect. Luồng dữ liệu như sau:

Ứng dụng ──ghi──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafka topic ──> Consumer

3. Dựng môi trường

3.1. Docker Compose

Bạn cần ba container: PostgreSQL bật logical replication, Kafka chạy chế độ KRaft và Kafka Connect có sẵn plugin Debezium.

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

Có ba điểm cần chú ý trong file này:

  • wal_level=logical là bắt buộc. Thiếu tham số này, Debezium không tạo được replication slot.
  • Kafka có hai listener. Container dùng kafka:29092, còn ứng dụng Spring Boot chạy trên máy bạn dùng localhost:9092.
  • Kafka Connect lưu cấu hình, offset và trạng thái vào ba topic riêng. Nhờ đó connector restart xong sẽ đọc tiếp từ vị trí cũ trong WAL.

Khởi động và tạo bảng:

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 yêu cầu PostgreSQL ghi toàn bộ dòng cũ vào WAL. Nếu không có lệnh này, trường before trong event UPDATE sẽ là null, còn event DELETE chỉ chứa khóa chính.

3.2. Đăng ký connector

Kafka Connect cung cấp REST API ở cổng 8083. Bạn đăng ký connector PostgreSQL như sau:

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

Ý nghĩa các thuộc tính quan trọng:

  • plugin.name: pgoutput dùng plugin decoding có sẵn của PostgreSQL 10+, nên bạn không phải cài thêm extension.
  • topic.prefix cùng schema và tên bảng tạo ra tên topic. Ở đây topic là shop.public.orders.
  • table.include.list giới hạn những bảng được capture.
  • decimal.handling.mode: string khiến cột NUMERIC được gửi dưới dạng chuỗi "149.90". Mặc định Debezium mã hóa giá trị này thành bytes base64, rất khó đọc.
  • schemas.enable: false bỏ phần schema đi kèm mỗi message JSON, nên payload gọn hơn.

Kiểm tra trạng thái 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. Cấu trúc change event

Thêm rồi cập nhật một đơn hàng:

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

Sau đó đọc topic bằng console consumer có sẵn trong image Kafka:

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

Event UPDATE trông như sau (đã rút gọn trường source):

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

Trường op cho biết loại thay đổi:

op Ý nghĩa before after
c INSERT null dòng mới
u UPDATE dòng cũ dòng mới
d DELETE dòng cũ null
r Đọc trong snapshot ban đầu null dòng hiện có

Khi xóa một dòng, Debezium gửi thêm một tombstone, tức message có cùng key nhưng value là null. Tombstone cho phép log compaction của Kafka dọn hẳn key đó. Consumer của bạn phải xử lý được value null.

5. Tiêu thụ event bằng Spring Boot

5.1. Dependency và cấu hình

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

Phiên bản của các dependency trên do spring-boot-starter-parent 3.x quản lý.

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

Ở đây consumer nhận value dạng String rồi tự parse. Cách này giúp bạn kiểm soát tombstone và lỗi JSON dễ hơn so với dùng JsonDeserializer.

5.2. Model và 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) bỏ qua trường source và những trường mới Debezium có thể thêm sau này. Jackson tự chuyển chuỗi "149.90" sang BigDecimal, nên bạn không bị mất độ chính xác như khi dùng 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);
        }
    }
}

Parser trả Optional.empty() cho tombstone, nên listener chỉ cần bỏ qua trường hợp này.

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) cho phép Spring truyền null vào method khi gặp tombstone. Nếu thiếu annotation này, Spring sẽ báo lỗi conversion. Nhánh "r" được gộp với "c" vì với consumer, một dòng có sẵn trong snapshot cũng giống một dòng vừa được tạo.

Chạy ứng dụng với dữ liệu ở mục 4, bạn sẽ thấy:

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

6. Kiểm thử parser

Logic parse là phần dễ hỏng nhất khi Debezium hoặc schema thay đổi. Bạn có thể test phần này mà không cần 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);
    }
}

Test đầu kiểm tra mapping tên cột customer_email và kiểu BigDecimal. Test thứ hai đảm bảo tombstone không làm listener lỗi. Test thứ ba xác nhận JSON hỏng được báo lỗi rõ ràng thay vì bị nuốt im lặng.

7. Lỗi hay gặp và khi nào không nên dùng

Lưu ý: Replication slot giữ WAL cho đến khi Debezium xác nhận đã đọc. Nếu connector dừng nhiều ngày, WAL sẽ phình to và có thể làm đầy ổ đĩa của database. Bạn cần giám sát pg_replication_slots và xóa slot khi gỡ connector: SELECT pg_drop_replication_slot('shop_slot');.

Một số lỗi khác bạn sẽ hay gặp:

  • Consumer không idempotent. Debezium và Kafka Connect đảm bảo at-least-once. Sau khi restart, cùng một event có thể đến hai lần. Hãy dùng id cùng với source.lsn để bỏ qua bản trùng, hoặc thiết kế thao tác dạng upsert.
  • Giá trị NUMERIC thành chuỗi base64 khó hiểu. Bạn quên cấu hình decimal.handling.mode. Tương tự, cột TIMESTAMP mặc định được gửi dưới dạng số microsecond kể từ epoch, nên cần tự chuyển đổi.
  • before luôn null. Bảng chưa đặt REPLICA IDENTITY FULL.
  • Snapshot ban đầu quá lâu. Với bảng lớn, snapshot.mode=initial sẽ đọc toàn bộ bảng trước khi stream. Hãy cân nhắc no_data nếu bạn chỉ cần thay đổi từ bây giờ, hoặc dùng incremental snapshot.
  • Rò rỉ dữ liệu nội bộ. Event phản ánh đúng cấu trúc bảng. Nếu đưa thẳng ra cho service khác, bạn đang công khai schema database. Với event nghiệp vụ, hãy kết hợp Debezium với Outbox pattern: ứng dụng ghi vào bảng outbox trong cùng transaction, và Debezium chỉ capture bảng đó.
  • Dùng tài khoản postgres trong production. Hãy tạo user riêng có quyền REPLICATION và quyền SELECT trên các bảng cần capture.

Bạn không nên dùng CDC khi:

  • Chỉ có một ứng dụng và nó tự phát event được trong cùng process. Khi đó transactional outbox với một poller đơn giản là đủ.
  • Bạn cần event mang ý nghĩa nghiệp vụ như “OrderPaid” thay vì “dòng X đổi cột status”. CDC thô chỉ biết dữ liệu thay đổi, không biết lý do.
  • Đội vận hành chưa sẵn sàng quản lý Kafka Connect, replication slot và việc tiến hóa schema.

8. Kết luận

Trong bài này, bạn đã dựng PostgreSQL với logical replication, Kafka và Debezium, rồi đăng ký connector capture bảng orders. Bạn cũng đã tìm hiểu cấu trúc change event với các trường before, after và op, sau đó viết consumer Spring Boot xử lý cả tombstone, kèm test cho phần parse.

CDC loại bỏ rủi ro dual-write vì database trở thành nguồn sự thật duy nhất. Đổi lại, bạn phải giám sát replication slot, viết consumer idempotent và cân nhắc dùng Outbox pattern khi cần event nghiệp vụ rõ ràng.

kafkadebeziumcdcpostgresqlspring-boot
Ứng dụng của mìnhAlterChụp ảnh thành tem, gom vào sổ temiOS · Android

Bài cùng chủ đề