PostgreSQL 커넥터는 두 부분으로 이루어진다[1].
pgoutput — PostgreSQL 10 부터 서버에 내장된 기본 플러그인. 별도 설치가 필요 없어 권장된다.decoderbufs — Protobuf 기반. Debezium 커뮤니티가 관리한다. 서버에 직접 설치해야 한다.
wal2json은 더 이상 지원 목록에 없다. 예전 문서에서plugin.name=wal2json을 쓰던 구성은pgoutput으로 바꾼다. 지원 PostgreSQL 버전은 14 · 15 · 16 · 17 · 18 이다[2].
pgoutput 은 REPLICA IDENTITY 설정에 따라 before 이미지가 달라진다. 기본키가 없는 테이블은 ALTER TABLE ... REPLICA IDENTITY FULL 을 주어야 update · delete 의 before 값이 온다.postgresql.conf 에서 논리 복제를 켜고 재시작한다.
wal_level = logical
max_wal_senders = 4
max_replication_slots = 4
# decoderbufs 를 쓸 때만
#shared_preload_libraries = 'decoderbufs'
복제 권한을 가진 역할을 만든다.
sudo systemctl restart postgresql-16
psql -U postgres
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD '${DEBEZIUM_PASSWORD}';
GRANT CONNECT ON DATABASE test TO debezium;
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
-- pgoutput 용 publication. 커넥터가 자동 생성하게 두려면 소유 권한이 필요하다.
CREATE PUBLICATION dbz_publication FOR ALL TABLES;
pg_hba.conf 에 복제 접속을 연다. 고친 뒤에는 다시 읽어야 한다 — pg_ctl reload 또는 SELECT pg_reload_conf();.
# TYPE DATABASE USER ADDRESS METHOD
host replication debezium 192.168.0.0/16 scram-sha-256
host test debezium 192.168.0.0/16 scram-sha-256
시험용 데이터베이스와 테이블을 만든다.
CREATE DATABASE test;
\c test
CREATE TABLE test_table (id char(10) NOT NULL, code char(10), PRIMARY KEY (id));
슬롯이 도는지 손으로 확인하려면 pg_recvlogical 을 쓴다(커넥터가 슬롯을 직접 만들므로 필수는 아니다).
# 슬롯 생성
pg_recvlogical -d test --slot test_slot --create-slot -P pgoutput
# 슬롯 조회
psql -d test -c "select slot_name, plugin, active, restart_lsn from pg_replication_slots;"
# 슬롯 제거
pg_recvlogical -d test --slot test_slot --drop-slot
cd /tmp
curl -LO https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/3.6.3.Final/debezium-connector-postgres-3.6.3.Final-plugin.tar.gz
tar -xzf debezium-connector-postgres-3.6.3.Final-plugin.tar.gz -C $KAFKA_HOME/plugins/
# plugins 디렉터리 바로 아래에 debezium-connector-postgres 디렉터리가 있어야 한다
ls $KAFKA_HOME/plugins/debezium-connector-postgres/
$KAFKA_HOME/config/register-postgres.properties. 2.0 부터 database.server.name 은 topic.prefix 다[1:1].
name=pg-conn
connector.class=io.debezium.connector.postgresql.PostgresConnector
tasks.max=1
plugin.name=pgoutput
database.hostname=${POSTGRES_HOST}
database.port=5432
database.user=debezium
database.password=${DEBEZIUM_PASSWORD}
database.dbname=test
# 토픽 이름 접두사. <topic.prefix>.<schema>.<table>
topic.prefix=pgserver
table.include.list=public.test_table
# 복제 슬롯·publication 이름
slot.name=debezium_pgserver
publication.name=dbz_publication
# 커넥터가 publication 을 만들지 여부 — all_tables | disabled | filtered
publication.autocreate.mode=filtered
Kafka Connect standalone 설정.
bootstrap.servers=${KAFKA_SERVER_IP_ADDRESS}:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/home/kafka/kafka/data/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000
# 플러그인 경로. 이 경로 바로 아래에 debezium 라이브러리 디렉터리가 있어야 한다
plugin.path=/home/kafka/kafka/plugins
cd $KAFKA_HOME
bin/connect-standalone.sh config/connect-standalone.properties config/register-postgres.properties
운영에서는 분산 모드로 띄우고 REST 로 등록한다.
curl -s -X POST -H 'Content-Type: application/json' http://connect-host:8083/connectors -d '{
"name": "pg-conn",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"plugin.name": "pgoutput",
"database.hostname": "db01.example.com",
"database.port": "5432",
"database.user": "debezium",
"database.password": "${DEBEZIUM_PASSWORD}",
"database.dbname": "test",
"topic.prefix": "pgserver",
"table.include.list": "public.test_table",
"slot.name": "debezium_pgserver"
}
}'
curl -s http://connect-host:8083/connectors/pg-conn/status
# PostgreSQL 쪽
psql -d test -c "insert into test_table values ('id1', 'code1');"
psql -d test -c "update test_table set code='11111' where id='id1';"
# Kafka 쪽
$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list | grep pgserver
$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic pgserver.public.test_table --from-beginning
로그 파일을 직접 볼 때는 덤프 도구를 쓴다.
$KAFKA_HOME/bin/kafka-dump-log.sh --print-data-log --files /home/kafka/logdata/pgserver.public.test_table-0/00000000000000000000.log
Debezium connector for PostgreSQL — 2026-09-20 확인. https://debezium.io/documentation/reference/stable/connectors/postgresql.html ↩︎ ↩︎
Debezium 3.6 release series — 지원 버전 표 — 2026-09-20 확인. https://debezium.io/releases/3.6/ ↩︎