Integrasi Database & CDC #

Menghubungkan database operasional relasional (RDBMS) dengan Apache Kafka merupakan salah satu skenario integrasi data yang paling sering ditemui di industri. Data transaksi yang masuk ke database utama harus segera disalurkan ke sistem hilir untuk diolah secara real-time. Ada dua cara utama untuk melakukan integrasi ini menggunakan Kafka Connect: pendekatan tradisional berbasis kueri polling menggunakan JDBC Source Connector, dan pendekatan modern berbasis transaksi menggunakan Change Data Capture (CDC). Meskipun pendekatan query polling terlihat mudah di awal, ia memiliki keterbatasan mendasar yang kritis untuk operasional skala besar. Artikel ini akan membedah kelemahan pendekatan query polling, menguraikan konsep dasar CDC berbasis log transaksi, menjelaskan arsitektur engine Debezium, serta mengupas penanganan evolusi skema (schema evolution) saat struktur tabel database berubah.

Batasan Pendekatan Query Polling Tradisional (JDBC Source Connector) #

JDBC Source Connector bekerja dengan cara mengirimkan kueri SQL SELECT secara berkala (misalnya setiap 5 detik) ke database target. Untuk mendeteksi adanya baris baru atau baris yang diperbarui, connector mengandalkan kolom penanda pada tabel, seperti kolom numerik yang terus meningkat (incrementing id) atau kolom penanda waktu pembaruan (timestamp).

Meskipun sederhana, pendekatan polling ini memiliki empat kelemahan desain yang kritis:

1. Tidak Dapat Mendeteksi Operasi Hapus (Delete) #

Ini adalah kelemahan terbesar JDBC Source. Ketika sebuah baris data dihapus dari database melalui perintah SQL DELETE, baris tersebut hilang dari tabel fisik. Karena baris tersebut sudah tidak ada, kueri SELECT periodik dari JDBC Connector tidak akan pernah mendeteksinya. Akibatnya, sistem hilir (downstream) yang membaca data dari Kafka tidak akan pernah tahu bahwa baris tersebut telah dihapus, sehingga menyebabkan ketidaksinkronan data (data drift).

2. Overhead Query yang Berat pada Database Utama #

Melakukan kueri SELECT secara terus-menerus ke tabel-tabel besar dengan indeks yang tidak optimal akan membebani CPU dan I/O database operasional. Hal ini dapat memperlambat transaksi bisnis utama pelanggan (application database contention).

3. Kehilangan Perubahan Sementara (Intermediate Updates) #

Jika sebuah baris data diperbarui beberapa kali di antara dua interval polling, JDBC Connector hanya akan menangkap status akhir baris tersebut pada saat kueri dijalankan. Seluruh riwayat perubahan sementara yang terjadi di antara interval polling tersebut akan hilang.

4. Ketergantungan pada Modifikasi Skema Sumber #

Pendekatan ini mewajibkan adanya kolom last_updated_timestamp yang konsisten di setiap tabel. Sering kali, tim pengembang terpaksa mengubah skema tabel warisan (legacy schema) dan membuat indeks tambahan hanya untuk mendukung kebutuhan polling ini.

Apa Itu Change Data Capture (CDC) Berbasis Log? #

Change Data Capture (CDC) berbasis log adalah teknik integrasi data modern yang membaca setiap perubahan data secara langsung dari log transaksi biner database.

Setiap database relasional modern memiliki log transaksi internal (seperti Write-Ahead Log / WAL di PostgreSQL, Binlog di MySQL, atau Redo Log di Oracle). Log ini ditulis secara berurutan (append-only) sebelum data ditulis ke tabel fisik untuk menjamin sifat durabilitas transaksi (ACID).

sequenceDiagram
    participant App as Aplikasi Bisnis
    participant DB as Engine Database
    participant Log as Transaction Log (WAL/Binlog)
    participant CDC as Debezium CDC Task
    participant Kafka as Kafka Broker

    App->>DB: Jalankan SQL (INSERT/UPDATE/DELETE)
    Note over DB: Lakukan validasi transaksi
    DB->>Log: Tulis perubahan ke Log (WAL/Binlog)
    DB-->>App: Konfirmasi Sukses Transaksi
    Note over DB: Tulis data ke tabel fisik secara asinkron
    
    CDC->>Log: Baca stream byte log transaksi (Non-blocking)
    Log-->>CDC: Kirim event perubahan biner
    Note over CDC: Konversi biner ke SourceRecord
    CDC->>Kafka: Publish event ke Kafka Topic

Mengapa CDC Berbasis Log Lebih Unggul? #

• Non-Blocking dan Sangat Ringan: CDC tidak mengirimkan kueri SELECT ke tabel database. Ia hanya membaca file log transaksi yang sudah ada di disk secara asinkron, sehingga overhead terhadap performa database operasional hampir mendekati nol.
• Menangkap Semua Kejadian (Termasuk DELETE): Setiap operasi DELETE ditulis secara jelas di log transaksi. CDC dapat menangkap event hapus ini dan mengirimkannya ke Kafka, biasanya berupa pesan khusus dengan nilai null (Tombstone) untuk memberi tahu sistem hilir agar menghapus data terkait.
• Menangkap Perubahan Sementara: Karena setiap penulisan transaksi dicatat di log secara berurutan, CDC menjamin tidak ada perubahan data sekecil apa pun yang terlewatkan.
• Tidak Memerlukan Perubahan Skema: Database tidak perlu menambahkan kolom timestamp atau indeks tambahan pada tabel sumber.

Detail Mekanisme Inisialisasi Awal (Initial Snapshotting) #

Ketika Debezium pertama kali dijalankan pada database produksi yang sudah beroperasi selama bertahun-tahun, ada jutaan baris data historis yang sudah tersimpan di tabel-tabel database. Log transaksi biner aktif (seperti WAL di PostgreSQL atau Binlog di MySQL) biasanya dikonfigurasi untuk dibersihkan secara otomatis setelah jangka waktu tertentu (misalnya 7 hari) untuk menghemat ruang disk.

Artinya, log transaksi aktif tidak lagi menyimpan seluruh sejarah data sejak database dibuat pertama kali. Oleh karena itu, Debezium memerlukan mekanisme khusus untuk menyelaraskan keadaan awal data melalui Initial Snapshotting.

Alur Kerja Snapshot Bertahap #

Secara default, Debezium menjalankan mode snapshot initial dengan urutan langkah yang sangat teratur:

1. Pemeriksaan Offset Terakhir: Debezium memeriksa apakah sudah ada koordinat offset terdaftar untuk database ini di topik connect-offsets. Jika sudah ada, proses snapshot dilewati, dan Debezium langsung beralih membaca log transaksi dari offset tersebut.
2. Locking (Sesuai Mode): Debezium melakukan locking pembacaan schema pada database sumber. Pada MySQL, Debezium secara default memperoleh global read lock (FLUSH TABLES WITH READ LOCK) untuk mengambil koordinat Binlog terbaru secara konsisten. Pada PostgreSQL, Debezium memanfaatkan mekanisme Exported Snapshot bawaan transaksi tingkat isolasi SERIALIZABLE untuk menghindari global locking yang mengganggu penulisan database.
3. Pembacaan Skema DDL: Debezium membaca seluruh skema struktur tabel yang masuk dalam cakupan integrasi dan mendaftarkannya ke Schema Registry.
4. Data Dumping (Pembacaan Massal): Debezium menjalankan kueri SELECT * FROM table secara efisien (menggunakan streaming result set) untuk menyalin keadaan akhir data dari seluruh tabel saat itu juga. Data ini dikirim ke topik Kafka dengan kode operasi op: r (read).
5. Pelepasan Lock dan Transisi ke Log: Setelah semua tabel disalin, kunci pembacaan dilepaskan. Debezium merekam koordinat log transaksi biner tempat ia memulai snapshot, lalu bertransisi secara mulus ke mode membaca stream biner secara real-time untuk memproses transaksi yang terjadi setelah snapshot dimulai.

Arsitektur Debezium sebagai CDC Engine Utama #

Debezium adalah kumpulan connector CDC open-source terkemuka yang dibangun di atas framework Kafka Connect. Debezium menyediakan plugin connector khusus untuk berbagai database populer seperti PostgreSQL, MySQL, SQL Server, Oracle, MongoDB, dan Cassandra.

Cara Kerja Debezium #

Debezium bertindak layaknya instansi replika dari database sumber. Sebagai contoh, Debezium MySQL Connector akan mendaftarkan dirinya ke server MySQL sebagai server slave replikasi menggunakan protokol replikasi biner MySQL. Server MySQL kemudian akan mengalirkan event perubahan log langsung ke Debezium secara real-time.

Anatomi Payload Event Debezium #

Setiap event perubahan data yang dikirim oleh Debezium ke Kafka memiliki struktur JSON/Avro terperinci yang mencakup keadaan sebelum (before) dan sesudah (after) data diubah, serta informasi metadata sumber (source).

Berikut adalah contoh riil isi payload JSON dari Debezium PostgreSQL Connector ketika terjadi operasi UPDATE pada tabel customers:

{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 102,
      "name": "Budi Santoso",
      "email": "[email protected]",
      "updated_at": 1672531190000
    },
    "after": {
      "id": 102,
      "name": "Budi Santoso",
      "email": "[email protected]",
      "updated_at": 1672531205000
    },
    "source": {
      "version": "2.1.2.Final",
      "connector": "postgresql",
      "name": "postgres-prod",
      "ts_ms": 1672531205120,
      "db": "inventory",
      "schema": "public",
      "table": "customers",
      "txId": 524,
      "lsn": 24598230
    },
    "op": "u",
    "ts_ms": 1672531205250
  }
}

Penjelasan Field Payload: #

• before: Keadaan baris data sebelum transaksi dijalankan. Berguna bagi konsumen yang ingin membandingkan nilai lama dengan nilai baru.
• after: Keadaan baris data setelah transaksi sukses dijalankan.
• source: Metadata database sumber yang sangat kaya, memuat nama database, nama tabel, ID transaksi internal (txId), timestamp transaksi di database (ts_ms), hingga log sequence number (lsn).
• op: Kode operasi yang memicu event ini. Nilainya adalah:
  • c (create) untuk operasi INSERT
  • u (update) untuk operasi UPDATE
  • d (delete) untuk operasi DELETE
  • r (read) untuk pembacaan inisiasi awal (snapshot).

Tuning Performa Pipeline Debezium CDC #

Untuk memastikan bahwa laju perubahan data di database utama dapat dialirkan tanpa latensi yang tinggi ke Kafka, kita perlu menyesuaikan beberapa parameter tuning performa penting di konfigurasi Debezium task:

# Jumlah maksimum event perubahan log yang dibaca dari antrean internal 
# dalam satu siklus pengiriman ke Kafka Connect. Default: 2048.
max.batch.size=4096

# Kapasitas maksimum antrean internal yang menampung pesan sebelum ditulis ke Kafka.
# Nilai ini harus selalu lebih besar dibanding max.batch.size (misal 2x atau 3x lipat).
max.queue.size=12288

# Durasi jeda maksimal (dalam milidetik) bagi thread poller Debezium 
# untuk menunggu event log transaksi baru masuk ke antrean sebelum dikirim.
poll.interval.ms=500

# Menentukan bagaimana tipe data angka presisi tinggi (Decimal) diproses.
# Setel ke 'double' atau 'string' agar tidak menghasilkan overhead biner BigDecimal Java yang rumit.
decimal.handling.mode=double

Penanganan Schema Evolution (Evolusi Skema) #

Dalam operasional jangka panjang, struktur tabel database pasti akan mengalami perubahan (DDL changes), seperti penambahan kolom baru, penghapusan kolom lama, atau modifikasi tipe data kolom. Proses penanganan perubahan ini di dalam pipa data terdistribusi disebut sebagai Schema Evolution.

Jika kita tidak mengelola evolusi skema dengan benar, modifikasi sederhana di database PostgreSQL dapat langsung memicu kegagalan berantai (cascade failure) pada seluruh konsumen hilir yang kaget menerima format data baru.

1. Integrasi Debezium dengan Schema Registry #

Untuk mengelola evolusi skema dengan aman, kita harus menggunakan Confluent Schema Registry bersama dengan format data serialisasi Avro or Protobuf.

Saat skema database berubah:

1. Debezium mendeteksi perubahan struktur DDL dari log transaksi.
2. Debezium merakit skema data internal baru.
3. Debezium mengirimkan skema baru tersebut ke Schema Registry untuk divalidasi.
4. Schema Registry memeriksa aturan kompatibilitas skema yang telah disetel. Jika valid, registry menetapkan ID skema baru.
5. Debezium mempublikasikan pesan biner baru ke Kafka dengan ID skema baru.

2. Memilih Aturan Kompatibilitas Skema (Compatibility Rules) #

Schema Registry menyediakan beberapa tingkat kompatibilitas yang menentukan bagaimana skema boleh berubah:

• BACKWARD Compatibility: Konsumen yang menggunakan skema baru dijamin tetap bisa membaca data yang ditulis dengan skema lama. Ini adalah mode default dan paling aman jika kita ingin meng-upgrade konsumen terlebih dahulu sebelum produser (Debezium).
  • Aturan: Kolom baru hanya boleh ditambahkan jika kolom tersebut bersifat opsional (memiliki nilai default atau boleh null). Kolom lama tidak boleh dihapus kecuali memiliki nilai default.
• FORWARD Compatibility: Konsumen yang menggunakan skema lama dijamin tetap bisa membaca data yang ditulis dengan skema baru.
  • Aturan: Kolom lama boleh dihapus. Kolom baru tidak boleh ditambahkan kecuali bertipe opsional.
• FULL Compatibility: Menjamin kompatibilitas dua arah (kombinasi Backward dan Forward).

Rekomendasi Konfigurasi Produksi Debezium #

Berikut adalah parameter penting yang harus kita setel untuk menjaga kestabilan skema di Debezium:

# Mengabaikan kolom DDL database yang tidak didukung oleh Kafka Connect
# agar task tidak langsung crash saat admin database menjalankan query DDL kustom
database.history.skip.unparseable.ddl=true

# Menyimpan riwayat perubahan DDL database ke topik Kafka khusus
# untuk pemulihan skema historis jika task dijalankan ulang dari awal
database.history.kafka.topic=schema-changes.inventory
database.history.kafka.bootstrap.servers=kafka-1:9092

Perbandingan JDBC Source vs CDC (Debezium) #

Untuk membantu kita memilih solusi integrasi database yang tepat, berikut adalah matriks komparasi fitur secara menyeluruh:

Ringkasan #

• Kelemahan Polling — JDBC Source Connector menggunakan query SELECT berkala yang membebani database operasional dan tidak mampu mendeteksi operasi penghapusan data (DELETE).
• Log-Based CDC — CDC membaca log transaksi biner database (seperti WAL atau Binlog) secara non-blocking, menawarkan overhead minimal dan merekam setiap detail transaksi.
• Debezium Payload — Payload Debezium sangat komprehensif, mencatat data sebelum (before), data sesudah (after), metadata sumber (source), serta tipe operasi (op).
• Initial Snapshotting — Mekanisme snapshot menyalin keadaan data awal secara aman dari database sebelum Debezium beralih memproses byte log transaksi aktif.
• Schema Evolution — Evolusi skema harus dikelola menggunakan Schema Registry untuk menjamin kompatibilitas format data saat struktur tabel database diubah.

  ← Sebelumnya: Standalone vs Distributed Mode   Berikutnya: Integrasi File & Object Storage →