Debezium과 Kafka를 사용한 Change Data Capture

1. 개요

한 서비스가 다른 데이터베이스의 데이터가 방금 변경되었는지 알아야 할 때, 일반적인 방식은 데이터베이스에 저장한 후 Kafka에 메시지를 보내는 것입니다. 이 방식을 dual-write라고 부르며 데이터 불일치를 유발하기 쉽습니다. Change Data Capture(CDC)는 데이터베이스의 트랜잭션 로그를 직접 읽음으로써 이 문제를 해결합니다.

이 글에서는 Docker Compose로 PostgreSQL, Kafka, Debezium을 구성하고 커넥터를 등록합니다. 그 후 Spring Boot consumer를 작성하여 orders 테이블의 INSERT, UPDATE, DELETE 이벤트를 처리합니다.

2. 왜 CDC가 필요한가

친숙한 dual-write 코드를 살펴봅시다:

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

이 코드는 두 가지 문제가 있습니다:

• save가 성공한 후 Kafka가 실패하면, 데이터베이스에는 주문이 있지만 이벤트가 전송되지 않습니다.
• send가 성공하지만 트랜잭션이 롤백되면, consumer는 존재하지 않는 주문의 이벤트를 받습니다.

CDC는 이 흐름을 역전시킵니다. 애플리케이션은 데이터베이스에만 기록합니다. 별도의 프로세스가 트랜잭션 로그(PostgreSQL의 경우 WAL, MySQL의 경우 binlog)를 읽고 커밋된 각 변경에 대한 이벤트를 발행합니다. 로그는 커밋된 변경만 포함하므로, 이벤트는 항상 실제 데이터와 일치합니다.

Debezium은 Kafka Connect에서 실행되는 CDC 커넥터 모음입니다. 데이터 흐름은 다음과 같습니다:

애플리케이션 ──기록──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafka topic ──> Consumer

3. 환경 구성

3.1. Docker Compose

세 개의 컨테이너가 필요합니다: 논리적 복제를 활성화한 PostgreSQL, KRaft 모드로 실행되는 Kafka, Debezium 플러그인이 포함된 Kafka Connect.

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

이 파일에서 주의해야 할 세 가지 사항:

• wal_level=logical은 필수입니다. 이 파라미터가 없으면 Debezium이 복제 슬롯을 생성할 수 없습니다.
• Kafka에는 두 개의 리스너가 있습니다. 컨테이너는 kafka:29092를 사용하고, 시스템에서 실행되는 Spring Boot 애플리케이션은 localhost:9092를 사용합니다.
• Kafka Connect는 구성, 오프셋, 상태를 세 개의 별도 토픽에 저장합니다. 따라서 커넥터가 재시작된 후 WAL의 이전 위치에서 계속 읽을 수 있습니다.

시작하고 테이블을 생성합니다:

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은 PostgreSQL이 이전 행 전체를 WAL에 기록하도록 요구합니다. 이 설정이 없으면 UPDATE 이벤트의 before 필드가 null이 되고, DELETE 이벤트는 기본 키만 포함합니다.

3.2. 커넥터 등록

Kafka Connect는 포트 8083에서 REST API를 제공합니다. PostgreSQL 커넥터를 다음과 같이 등록합니다:

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

주요 속성의 의미:

• plugin.name: pgoutput은 PostgreSQL 10+ 기본 제공 디코딩 플러그인을 사용하므로 추가 extension을 설치할 필요가 없습니다.
• topic.prefix는 스키마 및 테이블 이름과 함께 토픽 이름을 생성합니다. 여기서 토픽은 shop.public.orders입니다.
• table.include.list는 캡처할 테이블을 제한합니다.
• decimal.handling.mode: string은 NUMERIC 열을 "149.90" 문자열로 보냅니다. 기본적으로 Debezium은 이 값을 base64 바이트로 인코딩하여 읽기 어렵습니다.
• schemas.enable: false는 각 JSON 메시지와 함께 제공되는 schema 부분을 제거하므로 페이로드가 더 작아집니다.

커넥터 상태 확인:

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. Change event 구조

주문을 추가한 다음 업데이트합니다:

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

그 후 Kafka 이미지에 포함된 console consumer를 사용하여 토픽을 읽습니다:

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

UPDATE 이벤트는 다음과 같습니다(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
}

op 필드는 변경 유형을 나타냅니다:

op	의미	before	after
c	INSERT	null	새 행
u	UPDATE	이전 행	새 행
d	DELETE	이전 행	null
r	초기 스냅샷 읽기	null	현재 행

행을 삭제할 때 Debezium은 추가로 tombstone을 보냅니다. 즉, 같은 키를 가진 메시지이지만 값은 null입니다. Tombstone은 Kafka의 로그 압축이 해당 키를 완전히 제거하도록 합니다. consumer는 null 값을 처리할 수 있어야 합니다.

5. Spring Boot로 이벤트 처리

5.1. 의존성 및 구성

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

위의 의존성 버전은 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

여기서 consumer는 String 형식의 값을 받아서 직접 파싱합니다. 이 방식은 JsonDeserializer를 사용하는 것보다 tombstone과 JSON 오류를 더 쉽게 제어할 수 있습니다.

5.2. 모델 및 파서

@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)은 source 필드와 Debezium이 나중에 추가할 수 있는 새로운 필드를 무시합니다. Jackson은 자동으로 문자열 "149.90"을 BigDecimal로 변환하므로 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);
        }
    }
}

파서는 tombstone에 대해 Optional.empty()를 반환하므로 listener는 이 경우만 건너뛰면 됩니다.

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)는 tombstone을 만날 때 Spring이 null을 메서드에 전달하도록 허용합니다. 이 주석이 없으면 Spring은 변환 오류를 보고합니다. "r" 분기는 "c"와 병합됩니다. consumer 관점에서 스냅샷에 이미 존재하는 행은 방금 생성된 행과 같기 때문입니다.

섹션 4의 데이터로 애플리케이션을 실행하면 다음을 볼 수 있습니다:

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

6. 파서 테스트

파싱 로직은 Debezium 또는 스키마가 변경될 때 가장 깨지기 쉬운 부분입니다. 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);
    }
}

첫 번째 테스트는 customer_email 열 이름 매핑과 BigDecimal 타입을 확인합니다. 두 번째 테스트는 tombstone이 listener에 오류를 일으키지 않도록 합니다. 세 번째 테스트는 손상된 JSON이 조용히 무시되지 않고 명확한 오류를 생성하는지 확인합니다.

7. 자주 발생하는 오류와 CDC를 사용하지 말아야 할 경우

주의: 복제 슬롯은 Debezium이 읽었음을 확인할 때까지 WAL을 유지합니다. 커넥터가 며칠 동안 중지되면 WAL이 커져서 데이터베이스의 디스크를 가득 채울 수 있습니다. pg_replication_slots를 모니터링하고 커넥터를 제거할 때 슬롯을 삭제해야 합니다: SELECT pg_drop_replication_slot('shop_slot');.

발생할 수 있는 다른 오류들:

• Consumer가 멱등성이 아님. Debezium과 Kafka Connect는 at-least-once를 보장합니다. 재시작 후 동일한 이벤트가 두 번 올 수 있습니다. 중복을 건너뛰기 위해 id와 source.lsn을 함께 사용하거나, upsert 형태의 작업을 설계합니다.
• NUMERIC 값이 이해하기 어려운 base64 문자열로 표시됨. decimal.handling.mode 설정을 빠뜨렸습니다. 마찬가지로 TIMESTAMP 열은 기본적으로 epoch 이후의 마이크로초 수로 전송되므로 직접 변환해야 합니다.
• before가 항상 null임. 테이블에 REPLICA IDENTITY FULL이 설정되지 않았습니다.
• 초기 스냅샷이 너무 오래 걸림. 큰 테이블의 경우 snapshot.mode=initial은 스트림 시작 전에 전체 테이블을 읽습니다. 지금부터의 변경만 필요하면 no_data를 고려하거나, incremental snapshot을 사용합니다.
• 내부 데이터 누출. 이벤트는 테이블 구조를 정확히 반영합니다. 다른 서비스에 직접 전달하면 데이터베이스 스키마를 공개하는 것입니다. 비즈니스 이벤트의 경우 Debezium을 Outbox pattern과 결합합니다: 애플리케이션은 같은 트랜잭션에서 outbox 테이블에 기록하고, Debezium은 해당 테이블만 캡처합니다.
• 프로덕션에서 postgres 계정 사용. 복제 권한과 캡처할 테이블에 대한 SELECT 권한이 있는 별도 사용자를 만드세요.

CDC를 사용하지 말아야 할 경우:

• 애플리케이션이 하나이고 같은 프로세스 내에서 이벤트를 생성할 수 있을 때. 이 경우 transactional outbox와 간단한 poller로 충분합니다.
• “행 X의 상태 열 변경” 대신 “OrderPaid” 같은 비즈니스 의미를 가진 이벤트가 필요할 때. CDC는 데이터가 변경되었다는 것만 알 뿐 이유를 모릅니다.
• 운영 팀이 Kafka Connect, 복제 슬롯 및 스키마 진화를 관리할 준비가 되지 않았을 때.

8. 결론

이 글에서는 논리적 복제를 사용하여 PostgreSQL, Kafka, Debezium을 구성하고 orders 테이블을 캡처하는 커넥터를 등록했습니다. before, after, op 필드를 포함한 change event 구조를 배웠고, tombstone을 처리하는 Spring Boot consumer를 작성했으며, 파싱 부분에 대한 테스트를 작성했습니다.

CDC는 데이터베이스가 유일한 신뢰의 원천이 되기 때문에 dual-write의 위험을 제거합니다. 대신 복제 슬롯을 모니터링하고 멱등성 있는 consumer를 작성해야 하며, 명확한 비즈니스 이벤트가 필요한 경우 Outbox pattern 사용을 고려해야 합니다.