[Debezium] Debezium PostgreSQL Connector 동작 방식 & 간단한 구현 (With. Transactional Outbox Pattern)

0. 들어가기 전

이전 포스팅에서 Transactional Outbox Pattern에 대해서 다뤄보았었습니다.

https://ksh-coding.tistory.com/164

Transactional Outbox Pattern을 통해 Event Message 발행 보장하기

해당 포스팅에서 Transactional Outbox Pattern에 대한 이론을 살펴봤었습니다.

Transactional Outbox Pattern의 Flow를 도식화하면 다음과 같았습니다.

Transactional Outbox Flow

 

여기서 실제로 구현해야하는 부분은 크게 2가지로 나눌 수 있습니다.

• 비즈니스 로직에서 발생하는 이벤트 메타데이터를 Outbox 테이블에 저장
• Outbox 테이블의 이벤트 메타데이터를 기반으로 Kafka Message Relay (Send)

 

여기서 Outbox 테이블에 저장하는 부분은 단순 DB Insert이므로 생략하고,

이번 포스팅에서는 Debezium을 통해 Outbox 테이블의 데이터를 Kafka Broker로 Relay하는 부분을 간략하게 구현해보겠습니다.

 

Debezium은 여러 DBMS별 Kafka Connector를 제공하는데 프로젝트에서는 PostgreSQL을 사용했기 때문에

Debezium PostgreSQL Connector를 사용하여 구현해보겠습니다.

---

1. Debezium?

Debezium에 대해서 공식문서에서는 다음과 같이 소개하고 있습니다.

Debezium is a set of distributed services to capture changes in your databases so that your applications can see those changes and respond to them.

• 애플리케이션에서 DB의 데이터 변경을 알 수 있도록 DB의 변경을 캡쳐하는 분산 서비스

 

설명을 덧붙이면, Debezium은 잘 알려진 CDC(Change Data Capture) 라이브러리입니다.

• CDC란, DB에서 발생한 변경 내역을 추출해서 다른 시스템에 전달하는 기술입니다.

 

Debezium은 Apache Kafka 기반으로 구축되어 DBMS별로 Kafka Connector와 호환되는 Connector들을 제공합니다.

해당 Connector가 DB의 변경되는 데이터를 감지하고 기록, 해당 데이터를 Kafka Topic의 Event로 Streaming합니다.

 

 

이를 Transactional Outbox Pattern에 적용하면 다음과 같습니다.

• Debezium이 Outbox Table에 쌓이는 이벤트 데이터를 감지하여 해당 데이터를 Kafka Topic의 Event로 Streaming

---

2. Debezium(CDC)과 관련한 PostgreSQL 요소 

Debezium이 DB의 데이터 변경을 감지하는 만큼 DBMS별로 Connector가 별도로 존재합니다.

저는 DBMS로 PostgreSQL을 이용했기 때문에 PostgreSQL Connector에 대해 알아보겠습니다.

 

해당 Debezium PostgreSQL Connector의 동작 방식을 알아보기 전에,

Debezium PostgreSQL Connector의 동작 방식을 이해하기 위해서는 다음과 같은 PostgreSQL의 주요 개념을 이해해야 합니다.

• WAL (Write-Ahead Log)
• PostgreSQL Publication
• Logical Decoding Output Plug-in
• Logical Repliaction Slot
• WAL Sender

 

위 개념에 대해 간략하게 설명해보겠습니다.

 

 

2-1. WAL (Write-Ahead Log)

PostgreSQL의 모든 변경 사항은 WAL(Write-Ahead Log)에 쌓이게 됩니다.

 

WAL은 본질적으로는 PostgreSQL 데이터의 무결성을 보장하기 위한 요소이지만,

Logical Decoding 프로세스에서 WAL은 PostgreSQL의 데이터 변경 사항을 담는 저장소 역할을 합니다.

즉, Debezium Kafka Connector에서는 WAL에 쌓인 원장 데이터 변경 사항을 가져가는 것입니다.

 

 

 

2-2. PostgreSQL Publication

PostgreSQL의 Publication은 PostgreSQL에서 WAL의 데이터 중 외부에 내보낼 대상을 지정하는 PostgreSQL Object입니다.

Debezium에서 어떤 대상의 데이터 변경만 감지할 것인지를 필터링하는 역할을 수행합니다.

 

그래서 PostgreSQL의 모든 데이터 변경이 WAL에 쌓이지만 Debezium Kafka Connector는 Publication에서 지정한 데이터만을 읽어서 처리하게 됩니다.

즉, Debezium 입장에서 Publication은 데이터 변경을 감지할 대상을 필터링하는 역할을 수행합니다.

 

 

 

2-3. Logical Decoding Output Plug-in

Logical Decoding Output Plug-in은 PostgreSQL의 플러그인으로, 데이터 변경을 추출하는 역할을 수행합니다.

일반적으로 PostgreSQL 10 이상부터 내장되어 있는 'pgoutput'을 사용합니다.

 

해당 pgoutput은 PostgreSQL의 WAL 데이터 스트리밍을 위한 Logical Streaming Replication Protocol로 변환하는 역할을 합니다.

Debezium은 pgoutput을 통해 PostgreSQL 내부 포맷이 아닌 약속된 프로토콜로 변경 데이터를 처리할 수 있습니다.

 

 

 

2-4. Logical Replication Slot

Debezium Kafka Connector를 연결하기 위해 가장 중요한 요소는 Logical Replication Slot입니다.

Logical Replication Slot은 다음과 같은 역할을 수행합니다.

• Debezium에서 PostgreSQL의 변경 데이터를 받아가기 위한 연결 Channel
• 마지막에 읽은 WAL의 LSN (Log Sequence Number)를 저장하여 어디까지 읽었는지 저장

즉, Debezium은 PostgreSQL의 Logical Replication Slot을 지정하여 Subscribe Channel을 지정하고

해당 Logical Replication Slot에서 마지막으로 읽은 데이터 이후의 데이터를 스트리밍하여 처리하게 됩니다.

 

 

 

2-5. WAL Sender

WAL Sender는 Debezium에서 데이터 스트리밍 시 PostgreSQL에서 생성되는 내부 백그라운드 프로세스입니다.

해당 WAL Sender가 PG 내부의 데이터 스트리밍의 요청/응답을 처리하여 최종적으로 Debezium에게 전달합니다.

 

 

---

2. Debezium connector for PostgreSQL 

앞서서 PostgreSQL의 Debezium과 관련한 요소들을 간략하게 살펴봤었습니다.

해당 이해를 기반으로 Debezium Kafka Connector의 동작 Flow를 그림으로 나타내면 다음과 같습니다.

Debezium PostgreSQL Connector 동작 Flow

 

 

이를 동작 순서로 살펴보면 다음과 같습니다.

Debezium -> PostgreSQL 연결 (Access)

• 1. Debezium Kafka Application -> PostgreSQL 연결
• 2. PostgreSQL 내 데이터 스트리밍 백그라운드 프로세스인 WAL Sender 생성

 

Change Data Capture (CDC)

• 3. Debezium에서 지정한 Logical Replication Slot의 마지막 WAL의 LSN 조회
• 4. 마지막 WAL LSN 이후의 변경 데이터를 WAL에서 조회
• 5. 조회한 WAL 변경 데이터 중 Publication에서 지정한 데이터만 필터링
• 6. 변경 데이터를 pgoutput에서 Debezium에서 처리하기 위한 Logical Streaming Replication Protocol으로 변환
• 7. 최종 변경 데이터 Debezium Kafka Connector Application에 전달

 

Data Processing & Send

• 8. Debezium Kafka Application 내부에서 전달받은 데이터를 Kafka Message Format으로 가공 (커스텀 가능)
• 9. 가공된 Kafka Message를 Kafka Broker로 Send

 

Update Logical Replication Slot Last LSN

• 10. Kafka Message 전송이 정상적으로 완료되면 Debezium Kafka Application -> WAL Sender에 완료 메시지 전송
• 11. 완료 메시지를 받으면 Logical Replication Slot에 마지막 LSN Update

 

 

 

여기까지 Debezium PostgreSQL Connector의 동작 원리까지 살펴봤습니다.

간략하게 정리해봤지만 해당 원리를 이해하는 데까지 정말 많은 시간이 걸렸네요... 😂

---

3. Debezium 연동을 위한 PostgreSQL 설정

이제 PostgreSQL & Debezium PostgreSQL Connector & Kafka Broker를 연동해봅시다.

먼저, PostgreSQL 설정부터 살펴보면 앞서 주요 개념과 동작 Flow에서 봤듯이 다음과 같은 설정이 필요합니다.

• wal_level 변경
• Publication 생성
• Replication Slot 생성

 

3-1. PostgreSQL wal_level 변경

해당 PostgreSQL wal_level은 가장 중요한 Logical Replication Slot과 WAL을 연동하기 위한 설정입니다.

앞서 언급하지는 않았지만 Logical Repliaction을 사용하려면 wal_level을 logical로 설정해야 합니다.

 

PostgreSQL 10 이상부터는 기본 wal_level이 replica이므로 변경해야합니다.

로컬 환경에서는 도커를 사용해서 다음과 같이 커맨드를 지정했습니다.

일반적으로는 postgresql.conf 내의 wal_level을 수정해주면 됩니다.

services:
  pgsql:
    image: docker.io/library/postgres:15.5
    
    ...
    
    command: [
      'postgres','-c','wal_level=logical'
    ]
    
    ...

 

 

 

3-2. PostgreSQL Publication 생성

Publication 생성은 간단하게 다음과 같이 쿼리 1줄이면 생성할 수 있습니다.

더 자세한 상세 옵션들은 아래 PostgreSQL Publication 공식문서를 보면 확인할 수 있습니다.

https://www.postgresql.org/docs/current/sql-createpublication.html

 

CREATE PUBLICATION outbox FOR TABLE pay.outbox_pay
    WITH (
        publish = 'insert',
        publish_via_partition_root = true
    );

• CREATE PUBLICATION outbox : 'outbox'이라는 이름으로 PUBLICATION 생성
• FOR TABLE pay.outbox_pay : PUBLICATION 대상 테이블을 'pay.outbox_pay'으로 지정
  • 해당 부분이 앞서 동작 방식에서 살펴본 WAL을 필터링하는 부분입니다.
  • Publication에 지정되지 않은 테이블은 변경 데이터(WAL)가 외부(Debezium)으로 스트리밍되지 않게 됩니다.
• publish : 변경 이벤트를 발행할 DML 연산을 지정할 수 있습니다. (insert, update, delete, truncate)
  • Outbox 테이블의 Insert만 CDC 처리하기 때문에 insert로 지정
• publish_via_partition_root : 파티셔닝 테이블이 존재할 때, 해당 파티셔닝 테이블 데이터의 변경을 최상위 테이블 데이터 변경으로 발행
  • 현재 프로젝트에서 일자별 파티셔닝 테이블이 존재 (pay.outbox_pay_2025_06_05, pay.outbox_pay_2025_06_06, ...)
  • 이후 Debezium 설정 시 발행되는 테이블 명으로 커스텀할 때 상위 테이블로 커스텀되도록 발행을 상위 테이블로 하도록 true

 

 

 

 

3-3. PostgreSQL Logical Replication Slot 생성

Logical Replication Slot 생성도 간단하게 다음과 같이 PostgreSQL 함수를 사용하면 생성할 수 있습니다.

SELECT pg_create_logical_replication_slot(slot_name:='pay_local_outbox', plugin:='pgoutput');

• slot_name : 생성할 Logical Repliaction Slot 이름
• plugin : Slot에 적용할 plugin으로, pgoutput 지정

 

이렇게 Publication과 Replication Slot을 생성해주면 PostgreSQL 설정은 끝납니다.

---

4. Debezium PostgreSQL Connector 설정

이제 본격적인 Debezium PostgreSQL Connector 설정을 해봅시다.

해당 설정들은 아래 Debezium 공식문서의 PostgreSQL Connector 챕터에서 확인할 수 있습니다.

https://debezium.io/documentation/reference/stable/connectors/postgresql.html

Debezium connector for PostgreSQL :: Debezium Documentation

 

예시 설정들은 아주 간단한 학습용 설정으로 실제 프로덕션 환경에 적용해서 사용하기에는 무리가 있을 수 있습니다.

더 자세한 설정은 위의 문서를 참고하여 추가해나가시면 될 것 같습니다.

 

 

 

 

4-1. 설정 JSON 전체 내용

설정이 너무 방대하기 때문에 설정 JSON 파일 전체를 먼저 올리고 이후에 부분적으로 설명하겠습니다.

 

pay-outbox-source.json

 {
  "name": "outbox-source",
  "config": {
    // 1. Connector Class 지정
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

	// 2. PostgreSQL DB 정보 지정
    "database.hostname": "host.docker.internal",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "seongha12!@",
    "database.dbname": "pay_local",

	// 3. PostgreSQL Logical Replication Slot 설정
    "slot.name": "pay_local_outbox",

	// 4. PostgreSQL Publication 설정
    "publication.name": "outbox",
    "publication.autocreate.mode": "disabled",
    
    // 5. Logical Decoding Plugin 지정
    "plugin.name": "pgoutput",

	// 6. 메시지 처리할 Table 필터링
    "table.include.list": "outbox.outbox_pay",

	// 7. Kafka Topic 설정
    "topic.prefix": "pay",
    "topic.delimiter": ".",
    "topic.creation.enable": "true",
    "topic.creation.default.partitions": "1",
    "topic.creation.default.replication.factor": "1"
  }
}

 

 

1. Connector Class 지정

"connector.class": "io.debezium.connector.postgresql.PostgresConnector"

 

Debezium Connector로 사용할 커넥터 Class를 지정합니다.

PostgreSQL Connector는 위와 같은 Class로 지정하면 됩니다.

 

 

2. PostgreSQL DB 정보 설정

    "database.hostname": "host.docker.internal",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "seongha12!@",
    "database.dbname": "pay_local",

 

변경되는 데이터를 읽을 DB의 정보를 지정하는 부분입니다.

프로젝트 DB 정보를 입력해주면 됩니다.

 

참고로, 저는 로컬 환경에서 Kafka Connect (Debezium)과 PostgreSQL을 Docker Container로 구성했기 때문에

서로 다른 Docker Container의 접근을 위해 hostname으로 `host.docker.internal`를 사용했습니다.

 

 

3. PostgreSQL Logical Replication Slot 설정

    "slot.name": "pay_local_outbox"

 

앞서 생성한 Logical Repliaction Slot의 이름을 지정해주면 됩니다.

 

 

4. PostgreSQL Publication 설정

"publication.name": "outbox",
"publication.autocreate.mode": "disabled"

 

앞서 생성한 Publication 이름을 지정해주면 됩니다.

이때 `publication.autocreate.mode`는 Publication을 자동 생성하는 모드를 결정하는 옵션입니다.

• all_tables(default) : Publication이 있으면 해당 Publication 적용, 없다면 모든 테이블 대상으로 Publication 생성
• filtered : 커넥터 설정에 설정한 테이블(table.include.list, schema.include.list, ...)에 따라서 Publication 생성
• no_tables : Publication 생성 시 테이블을 지정하지 않고 생성 (모든 변경 데이터 패스)
• disabled : 해당 자동 생성 옵션 삭제, PG에서 직접 생성해야함.

 

위와 같은 여러 자동 생성 모드가 존재하지만, 저는 Publication은 PostgreSQL의 Object이기 때문에 

생성 역할이 Debezium Connector로 넘어가는 것은 좋지 않다고 판단했습니다.

그래서 disabled를 사용하고 직접 위에서 생성한 것처럼 PostgreSQL 쿼리로 생성하는 방식을 택했습니다.

 

 

5. PostgreSQL Logical Decoding Plug-in 설정

    "plugin.name": "pgoutput",

 

PostgreSQL의 Logical Decoding Plug In을 지정하는 옵션입니다.

PostgreSQL 10+ 버전에서는 기본이 pgoutput이므로 pgoutput으로 지정합니다.

 

 

 

6. 메시지를 처리할 Table 필터링

    "table.include.list": "outbox.outbox_pay",

 

앞서 Publication에서 WAL을 필터링하여 다른 테이블의 변경 이벤트가 스트리밍 되지는 않지만,

커넥터 설정에서도 필터링하는 테이블을 명확히 하기 위해 명시적으로 선언했습니다.

 

 

 

7. Kafka Topic 설정

    "topic.prefix": "pay",
    "topic.delimiter": ".",
    "topic.creation.enable": "true",
    "topic.creation.default.partitions": "1",
    "topic.creation.default.replication.factor": "1"

 

Kafka Broker에 Send할 Message의 Topic을 지정하는 부분입니다.

Debezium Connector가 자동으로 생성하는 Kafka Topic은 다음과 같은 기준으로 생성됩니다.

• topic.prefix + topic.delimiter + Schema Name + topic.delimiter + Table Name
  • topic.delimiter의 default : '.'
  • topic.prefix는 default X

따라서, `pay.outbox_pay`의 경우에는 다음과 같이 생성됩니다.

• pay.pay.outbox_pay

---

4. Debezium PostgreSQL Connector 생성

앞서 Connector 설정이 완료되었다면, 이제 해당 설정을 기반으로 Connector를 생성할 수 있습니다.

 

Kafka Connector는 Connector를 관리하는 REST API를 제공하여 REST API 호출으로 간단하게 커넥터를 관리할 수 있습니다.

관련 REST API는 다음과 같은 Confluent의 Kafka Connector 부분 공식문서에서 확인할 수 있습니다.

https://docs.confluent.io/platform/current/connect/references/restapi.html#connectors

Kafka Connect REST Interface for Confluent Platform | Confluent Documentation

 

 

REST API 중 Connector를 생성하는 API인 `POST /connectors`를 호출하여 생성하면 됩니다.

저는 IntelliJ의 http 파일을 사용하여 해당 폴더에 JSON을 위치시키고 다음과 같이 실행했습니다.

### 샘플 커넥터 생성 (Kafka Connector -> localhost:8083)
POST localhost:8083/connectors
Content-Type: application/json

< ./pay-outbox-source.json

 

해당 호출을 하게 되면 201 응답과 함께 커넥터가 생성된 것을 확인할 수 있습니다.

저는 Kafka UI를 Docker Container로 실행했기 때문에 다음과 같이 UI로도 확인할 수 있었습니다.

---

5. CDC 동작 테스트

이제 커넥터도 생성되었으므로 실제 Outbox 테이블에 쌓이는 변경 데이터를 Connector가 읽어가는지 확인해봅시다.

확인할 사항은 다음과 같습니다.

• 지정한 토픽(자동 생성 토픽)이 생성되었는지, 해당 토픽에 Outbox 테이블에 해당하는 메시지가 쌓이는지

 

이를 위해 임의로 Outbox 테이블에 데이터를 다음과 같이 2건 Insert 하여 이벤트 발행 상황을 구현했습니다.

insert into outbox.outbox
    (topic, event_id, created_at, event_type, aggregate_type, aggregate_id, metadata, payload)

values
    ('job', 'b77e5423-9ac0-43db-afb4-17c4b4675fb4', '2025-06-21 14:32:48.013000 +00:00', ...),
    ('job', 'bdf89548-cf3b-4e42-8092-caec5883f148', '2025-06-21 14:35:41.704000 +00:00', ...);

 

 

Kafka UI에서 확인한 결과, 아래와 같이 메시지가 정상적으로 지정한 토픽에 쌓이는 것을 확인할 수 있었습니다.

---

여기까지 해서 Debezium PostgreSQL Connector의 동작 방식과 간단한 구현까지 알아보았습니다.

사실 해당 구현만으로는 프로덕션에 Debezium Connector를 사용하기에는 무리가 있을 것입니다.

 

프로덕션에 사용하려면 쌓이는 메시지들을 프로젝트에 맞게 커스텀 Transformation을 하고 여러 추가 설정들이 필요합니다.

해당 설정들까지 다루기에는 너무 글이 방대해져서 동작 방식과 간단한 구현으로 글을 마무리하고자 합니다.

 

나머지 추가 설정들은 위에서 언급한 공식 문서들을 살펴보면 모두 나와 있으니 커스텀하여 사용하면 될 듯 합니다! 😀