Arsitektur data modern menuntut integrasi data bernilai bisnis tinggi secara real-time dari database transaksional (OLTP) ke sistem analit...
Arsitektur data modern menuntut integrasi data bernilai bisnis tinggi secara real-time dari database transaksional (OLTP) ke sistem analitik hilir, data lakehouse, dan microservices. Pendekatan integrasi data konvensional berbasis polling query periodik (SELECT * WHERE updated_at > ...) memiliki kelemahan mendasar: menimbulkan lonjakan latensi, risiko query locking, dan pemborosan CPU pada database sumber. Change Data Capture (CDC) berbasis log transaksi menawarkan pendekatan efisien dengan membaca mutasi data langsung dari Write-Ahead Log (WAL) PostgreSQL, meminimalkan query locking kendati proses decoding WAL tetap membutuhkan alokasi CPU, I/O, dan kapasitas disk server.
Dengan transisi ekosistem Apache Kafka menuju mode KRaft (Kafka Raft Metadata mode) yang mengeliminasi dependensi Apache ZooKeeper, arsitektur streaming kini menjadi jauh lebih ramping dan mudah dikelola. Artikel ini menyajikan panduan teknis langkah demi langkah untuk membangun pipeline CDC real-time dari PostgreSQL 16 menuju kluster Apache Kafka 3.8 (KRaft mode) menggunakan Debezium Connector 2.7 pada lingkungan lab pengujian terisolasi.
1. Konsep Arsitektur CDC Berbasis WAL dan KRaft Mode
Sebelum masuk ke tahap konfigurasi, penting untuk memahami bagaimana data bermutasi dan mengalir melalui lapisan-lapisan pipeline:
[Aplikasi OLTP] --(DML SQL)--> [PostgreSQL 16 Engine] --(Tulis WAL)--> [pgoutput Plugin] --(Logical Decoding)--> [Replication Slot] --(Streaming)--> [Debezium 2.7 Connect] --(JSON/Avro Envelope)--> [Kafka KRaft Broker] --(Internal 29092)--> [Consumer Container]
- PostgreSQL Write-Ahead Logging (WAL): Setiap transaksi INSERT, UPDATE, atau DELETE dicatat ke file WAL sebelum ditulis ke tabel data. Pada tingkat
wal_level = logical, informasi baris data disimpan secara terperinci sehingga dapat didekode menjadi aliran perubahan logis. - Plugin pgoutput: Plugin bawaan PostgreSQL sejak versi 10 yang mendekode perubahan WAL langsung ke format streaming standar tanpa memerlukan kompilasi ekstensi pihak ketiga.
- Debezium Connect 2.7: Menjalankan task Kafka Connect yang mengonsumsi stream dari replication slot PostgreSQL dan menerjemahkannya menjadi Kafka event messages.
- Apache Kafka KRaft Mode: Menghilangkan kebutuhan kluster eksternal ZooKeeper. Konsensus metadata kluster, penugasan partisi, dan pemilihan controller dikelola secara internal menggunakan protokol konsensus Raft terdistribusi.
2. Konfigurasi Sumber Data (PostgreSQL 16)
Langkah awal adalah mengonfigurasi mesin PostgreSQL agar mendukung logical replication, mengatur autentikasi jaringan, serta menerapkan prinsip least-privilege untuk user CDC.
2.1 Parameter postgresql.conf
Buka berkas postgresql.conf dan pastikan parameter replikasi logis berikut disetel:
# Mengaktifkan level informasi logis pada Write-Ahead Log
wal_level = logical
# Jumlah maksimum replication slot yang dapat dialokasikan (sesuaikan kebutuhan konektor)
max_replication_slots = 5
# Jumlah maksimum proses walsender yang berjalan bersamaan
max_wal_senders = 5
# Batas retensi WAL yang ditahan oleh replication slot sebelum di-invalidasi
# CATATAN: Parameter ini bertindak sebagai safety-valve per-slot dan BUKAN jaminan mutlak anti-disk full
max_slot_wal_keep_size = 8192MBParameter max_slot_wal_keep_size menetapkan ambang batas akumulasi WAL oleh suatu replication slot tertentu. Namun, parameter ini bukan hard cap kapasitas disk server dan bukan jaminan mutlak mencegah kehabisan kapasitas disk (disk exhaustion). Jika laju penulisan transaksi sangat tinggi sebelum checkpoint dieksekusi, atau bila proses archiving WAL eksternal (archive_command) mengalami kemacetan, partisi disk tetap berisiko penuh. Selain itu, jika batas ini terlampaui, PostgreSQL akan membatalkan slot (wal_status = 'lost') demi menyelamatkan ketersediaan database, yang berakibat konektor Debezium gagal dan menuntut prosedur snapshot ulang.
2.2 Konfigurasi Izin Akses Jaringan pada pg_hba.conf
PostgreSQL membedakan koneksi database standar (untuk query relasional katalog skema) dengan koneksi streaming replikasi melalui walsender. Pada berkas pg_hba.conf, kata kunci replication pada kolom database merupakan kata kunci protokol khusus tingkat kluster fisik untuk walsender, bukan nama database relasional logis. Debezium membutuhkan kedua jenis izin akses tersebut:
# /etc/postgresql/16/main/pg_hba.conf
# 1. Izinkan koneksi relasional reguler ke database target untuk query katalog metadata
host enterprise_db cdc_debezium <CIDR_SUBNET_LAB> scram-sha-256
# 2. Izinkan koneksi protokol streaming replikasi fisik/kluster (walsender)
host replication cdc_debezium <CIDR_SUBNET_LAB> scram-sha-256Nilai <CIDR_SUBNET_LAB> (misalnya 172.18.0.0/16 pada bridge Docker tertentu) bukanlah subnet universal. Subnet dan gateway alokasi Docker berbeda di setiap mesin host tergantung pada urutan pembuatan network dan konfigurasi daemon. Pastikan untuk memeriksa CIDR bridge aktual Anda (misalnya via docker network inspect bridge) atau gunakan IP host tepercaya.
Setelah mengedit kedua berkas tersebut, muat ulang konfigurasi PostgreSQL:
sudo systemctl restart postgresql2.3 Urutan DDL Tabel, Hak Akses (Least Privilege), dan Publikasi
Untuk menghindari kegagalan otorisasi, tabel target harus dibuat terlebih dahulu sebelum hak akses (GRANT) diberikan. Selanjutnya, gunakan ALTER DEFAULT PRIVILEGES secara eksplisit untuk role pemilik objek agar tabel masa depan otomatis dapat diakses oleh user CDC.
-- Masuk ke database target: enterprise_db
\c enterprise_db
-- 1. Buat role cdc_debezium dengan atribut REPLICATION
-- PENTING: Ganti placeholder <POSTGRES_CDC_PASSWORD> dengan kredensial rahasia yang kuat
CREATE USER cdc_debezium WITH REPLICATION PASSWORD '<POSTGRES_CDC_PASSWORD>';
GRANT CONNECT ON DATABASE enterprise_db TO cdc_debezium;
-- 2. Buat tabel transaksi TERLEBIH DAHULU sebelum mengeksekusi perintah GRANT
CREATE TABLE IF NOT EXISTS public.customer_transactions (
id SERIAL PRIMARY KEY,
customer_id INT NOT NULL,
amount NUMERIC(12, 2) NOT NULL,
status VARCHAR(30) DEFAULT 'PENDING',
created_at TIMESTAMPTZ DEFAULT clock_timestamp()
);
-- Atur REPLICA IDENTITY FULL agar WAL mencatat nilai baris lengkap sebelum mutasi (before) saat UPDATE/DELETE
ALTER TABLE public.customer_transactions REPLICA IDENTITY FULL;
-- 3. Berikan izin baca (SELECT) pada skema dan tabel yang telah dibuat
GRANT USAGE ON SCHEMA public TO cdc_debezium;
GRANT SELECT ON TABLE public.customer_transactions TO cdc_debezium;
-- 4. Atur ALTER DEFAULT PRIVILEGES untuk role postgres (atau role pemilik tabel)
-- agar tabel baru yang dibuat di masa mendatang otomatis memberikan izin SELECT ke cdc_debezium
ALTER DEFAULT PRIVILEGES FOR ROLE postgres IN SCHEMA public
GRANT SELECT ON TABLES TO cdc_debezium;
-- 5. Buat publikasi replikasi secara eksplisit (Best Practice Least-Privilege)
CREATE PUBLICATION debezium_pg_pub FOR TABLE public.customer_transactions;3. Penerapan Kafka KRaft dan Debezium Connect (Docker Lab Terisolasi)
Pada arsitektur lab terisolasi ini, kita menjalankan broker Apache Kafka dengan controller internal berbasis KRaft dan kontainer Debezium Connect.
Konfigurasi di bawah ini dirancang khusus untuk lingkungan lab pengujian lokal terisolasi. Port host di-bind secara eksplisit ke antarmuka loopback 127.0.0.1 agar tidak terekspos ke jaringan publik. Protokol PLAINTEXT dan HTTP REST tanpa enkripsi tidak boleh digunakan di lingkungan produksi multi-tenant. Lingkungan produksi wajib menerapkan enkripsi TLS/mTLS, autentikasi SASL/SCRAM, serta pembatasan akses firewall/reverse-proxy pada REST API Kafka Connect.
3.1 Berkas docker-compose.yml
Simpan konfigurasi berikut sebagai docker-compose.yml:
version: '3.8'
services:
kafka:
image: apache/kafka:3.8.0
container_name: kafka-kraft
ports:
- "127.0.0.1:9092:9092" # Loopback host binding untuk pengujian klien lokal
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_LOG_RETENTION_HOURS: 168
volumes:
- kafka-data:/var/lib/kafka/data
restart: unless-stopped
debezium:
image: debezium/connect:2.7.3.Final
container_name: debezium-connect
depends_on:
- kafka
ports:
- "127.0.0.1:8083:8083" # Loopback host binding untuk Kafka Connect REST API
environment:
BOOTSTRAP_SERVERS: kafka:29092
GROUP_ID: 1
CONFIG_STORAGE_TOPIC: _connect_configs
OFFSET_STORAGE_TOPIC: _connect_offsets
STATUS_STORAGE_TOPIC: _connect_statuses
CONFIG_STORAGE_REPLICATION_FACTOR: 1
OFFSET_STORAGE_REPLICATION_FACTOR: 1
STATUS_STORAGE_REPLICATION_FACTOR: 1
KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
KEY_CONVERTER_SCHEMAS_ENABLE: "false"
VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
volumes:
- debezium-data:/kafka/data
restart: unless-stopped
volumes:
kafka-data:
debezium-data:Jalankan stack kontainer lab dan verifikasi kesiapan endpoint Kafka Connect:
# Menjalankan kontainer di latar belakang
docker compose up -d
# Periksa status kesiapan endpoint REST Kafka Connect (port 8083 pada loopback)
curl -s http://127.0.0.1:8083/ | jq .4. Mendaftarkan Debezium PostgreSQL Connector
Pendaftaran konektor dilakukan dengan mengirimkan payload JSON melalui HTTP POST ke REST API Kafka Connect. Debezium akan terhubung ke PostgreSQL, memvalidasi publication, dan membuat replication slot bernama debezium_pg_slot.
4.1 Registrasi Payload JSON Connector melalui REST API
curl -X POST -H "Content-Type: application/json" --data '{
"name": "pg-transactions-cdc",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"plugin.name": "pgoutput",
"database.hostname": "<DOCKER_GATEWAY_OR_HOST_IP>",
"database.port": "5432",
"database.user": "cdc_debezium",
"database.password": "<POSTGRES_CDC_PASSWORD>",
"database.dbname": "enterprise_db",
"topic.prefix": "cdc_prod",
"table.include.list": "public.customer_transactions",
"slot.name": "debezium_pg_slot",
"publication.name": "debezium_pg_pub",
"publication.autocreate.mode": "disabled",
"decimal.handling.mode": "string",
"tombstones.on.delete": "false"
}
}' http://127.0.0.1:8083/connectorsdatabase.hostname: Masukkan alamat IP gateway antarmuka Docker bridge atau IP host PostgreSQL yang dapat dijangkau dari dalam kontainer. Perlu diingat bahwa alamat gateway subnet bukan universal; periksa rute gateway aktual mesin Anda (misal viaip routeataudocker network inspect).tasks.max: PostgreSQL logical decoding dibatasi 1 task per replication slot untuk menjamin urutan event mutasi data.publication.autocreate.mode: Disetel kedisabledkarena publikasidebezium_pg_pubtelah dikelola dan dibuat secara eksplisit oleh DBA.decimal.handling.mode: Disetel kestring. Nilaidoubledapat memicu ketidaktepatan presisi floating-point pada kalkulasi moneter. Opsistringmenjaga integritas angka desimal tanpa pembulatan tidak terduga.
4.2 Verifikasi Status Kesehatan Konektor
Pastikan konektor dan task berada dalam status RUNNING:
# Periksa status detail worker dan task konektor
curl -s http://127.0.0.1:8083/connectors/pg-transactions-cdc/status | jq .5. Verifikasi Replikasi Event Mutasi Data
Lakukan simulasi manipulasi data pada database sumber untuk menguji tangkapan event secara komprehensif pada operasi INSERT, UPDATE, dan DELETE.
5.1 Uji Simulasi DML (INSERT, UPDATE, DELETE)
Jalankan transaksi pengujian di PostgreSQL:
-- 1. Eksekusi transaksi INSERT (Pembuatan baris baru)
INSERT INTO public.customer_transactions (customer_id, amount, status)
VALUES (101, 750000.00, 'PAID');
-- 2. Eksekusi transaksi UPDATE (Modifikasi baris yang ada)
UPDATE public.customer_transactions
SET status = 'COMPLETED', amount = 745000.00
WHERE customer_id = 101;
-- 3. Eksekusi transaksi DELETE (Penghapusan baris)
DELETE FROM public.customer_transactions
WHERE customer_id = 101;5.2 Konsumsi Topik Kafka
Untuk mengonsumsi event streaming dari dalam jaringan kontainer Docker, consumer wajib menggunakan listener internal kafka:29092:
# Konsumsi topik dari dalam kontainer Kafka (menggunakan listener internal kafka:29092)
docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka:29092 \
--topic cdc_prod.public.customer_transactions \
--from-beginning
# Catatan: Bila mengonsumsi dari aplikasi klien pada host lokal mesin (luar kontainer),
# gunakan antarmuka loopback yang diekspos: --bootstrap-server 127.0.0.1:90925.3 Struktur Envelope JSON Debezium
Debezium menghasilkan struktur envelope standar yang memuat status sebelum mutasi (before), status setelah mutasi (after), identifier operasi (op), serta metadata posisi log transaksi (lsn, txId). Karakteristik representasi tiap operasi mutasi adalah sebagai berikut:
- Operasi Create (INSERT): Menghasilkan
op: "c". Blokbeforebernilainull, dan blokafterberisi data baris baru yang di-insert. - Operasi Update (UPDATE): Menghasilkan
op: "u". Blokbeforememuat nilai baris lama (tersedia lengkap berkat pengaturanREPLICA IDENTITY FULL), dan blokaftermemuat nilai baris terbaru setelah pembaruan. - Operasi Delete (DELETE): Menghasilkan
op: "d". Blokbeforememuat data baris terakhir sebelum dihapus, dan blokafterbernilainull.
Contoh struktur payload envelope JSON pada operasi UPDATE (dengan nilai skema sederhana tanpa converter schemas):
{
"before": {
"id": 1,
"customer_id": 101,
"amount": "750000.00",
"status": "PAID",
"created_at": 1700000000000000
},
"after": {
"id": 1,
"customer_id": 101,
"amount": "745000.00",
"status": "COMPLETED",
"created_at": 1700000000000000
},
"source": {
"version": "2.7.3.Final",
"connector": "postgresql",
"name": "cdc_prod",
"ts_ms": 1700000010000,
"db": "enterprise_db",
"schema": "public",
"table": "customer_transactions",
"txId": 1002,
"lsn": 248902848
},
"op": "u",
"ts_ms": 1700000010050
}5.4 Mitigasi Bahaya WAL Bloat dan Disk Exhaustion
Replication slot di PostgreSQL menahan segmen file WAL di disk agar tidak dihapus hingga Debezium mengonfirmasi (ACK) posisi LSN yang telah berhasil diproses. Jika Debezium terhenti berhari-hari sementara transaksi database terus berjalan masif, direktori pg_wal dapat membengkak hingga menghabiskan seluruh sisa kapasitas disk server.
Pantau kesehatan slot replikasi secara periodik dengan query diagnostik berikut:
SELECT
slot_name,
plugin,
active,
wal_status,
pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal_bytes
FROM pg_replication_slots;Jika kolom wal_status menunjukkan nilai extended atau unreserved, segera investigasi konektivitas Kafka Connect sebelum slot ter-invalidasi menjadi lost.
6. Hardening Keamanan Produksi dan Batasan Panduan
Sebelum menerapkan pipeline CDC ke lingkungan produksi multi-tenant, terapkan tiga pilar perlindungan operasional berikut:
6.1 Masking Data Sensitif (PII) dengan Single Message Transforms (SMT)
Hindari mengalirkan atribut sensitif (seperti data finansial rahasia atau identitas pribadi) ke broker Kafka secara mentah. Konfigurasikan SMT org.apache.kafka.connect.transforms.MaskField pada konfigurasi connector untuk menutupi atribut sensitif sebelum payload event meninggalkan memori worker.
6.2 Proteksi Kredensial dan Enkripsi Jaringan
Hindari menyertakan password secara plain-text dalam payload cURL. Manfaatkan framework Kafka Connect ConfigProvider (seperti DirectoryConfigProvider) untuk menginjeksikan variabel kredensial dari file berizin terbatas atau secrets management vault. Aktifkan enkripsi in-transit TLS dan autentikasi SASL SCRAM-SHA-512 untuk komunikasi jaringan non-localhost.
6.3 Isolasi Network Endpoint Kafka Connect REST API
Port 8083 mengekspos endpoint manajemen konfigurasi konektor. Batasi akses port 8083 hanya pada subnet administratif internal atau reverse proxy dengan proteksi autentikasi ketat agar pihak tidak berwenang tidak dapat mengubah konfigurasi koneksi database.
6.4 Ruang Lingkup dan Batasan Panduan
Catatan Transparansi Evaluasi: Konfigurasi dan sintaks logical replication pada artikel ini disusun mengacu pada dokumentasi resmi PostgreSQL 16, Debezium 2.7, dan Apache Kafka 3.8. Panduan ini berfokus pada arsitektur integrasi dasar dalam lingkungan lab pengujian lokal terisolasi. Metrik throughput throughput maksimum pesan per detik, latensi under-load, dan overhead CPU riil belum diuji secara empiris dalam panduan ini karena performa sangat bergantung pada karakteristik beban kerja (workload), spesifikasi storage I/O, serta kapasitas jaringan kluster. Selalu lakukan pengujian beban (load testing) mandiri pada lingkungan staging sebelum deployment produksi.
Referensi dan Sumber Resmi
- PostgreSQL 16 Documentation - Chapter 31: Logical Replication Architecture
- PostgreSQL 16 Documentation - Chapter 21.1: The pg_hba.conf File & Walsender Protocol
- PostgreSQL 16 Documentation - Server Configuration: Replication & WAL Retention
- Debezium 2.7 Documentation - PostgreSQL Connector Configuration Reference
- Debezium 2.7 Documentation - Debezium Architecture on Kafka Connect
- Apache Kafka 3.8 Documentation - KRaft (Kafka Raft Metadata Mode) Architecture
COMMENTS