Belajar Debezium - Instalasi & Menyiapkan Connector

#Pendahuluan

Di episode 0 kalian sudah menyalakan Kafka dan Kafka Connect dalam mode KRaft. Episode 3 ini memakai runtime tersebut untuk hal yang sesungguhnya: mendaftarkan connector pertama. Kalian akan menambahkan database sumber, membuat user khusus dengan hak minimal, mendaftarkan connector lewat REST API, dan membaca event CDC pertama di terminal.

Meskipun contoh utamanya MySQL, bagian akhir episode ini merangkum konfigurasi untuk PostgreSQL, MongoDB, dan SQL Server. Dengan satu pola kerja yang jelas, kalian bisa menambah sumber data baru hanya dengan menyalin dan menyesuaikan konfigurasi.

#Menyiapkan Database Sumber dengan Docker

Tambah service MySQL ke dalam docker-compose.yml yang sudah ada. Debezium menyediakan image contoh yang berisi database inventory lengkap dengan user debezium yang sudah punya privilege yang benar:

  mysql:
    image: quay.io/debezium/example-mysql:3.0
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: dbz
    volumes:
      - mysql_data:/var/lib/mysql
 
volumes:
  mysql_data:

Jalankan ulang stack agar service MySQL ikut terangkat:

docker compose up -d
docker compose ps

Pastikan ketiga container — kafka, connect, dan mysql — berada dalam status running sebelum lanjut.

#Mendaftarkan Connector MySQL

Connector didaftarkan melalui REST API Kafka Connect pada port 8083. Konfigurasi dikirim sebagai dokumen JSON melalui endpoint /connectors:

curl -i -X POST -H "Accept: application/json" -H "Content-Type: application/json" \
  http://localhost:8083/connectors/ -d '{
    "name": "inventory-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.server.id": "223344",
      "topic.prefix": "dbserver1",
      "database.include.list": "inventory",
      "schema.history.internal.kafka.topic": "schema-changes.inventory",
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092"
    }
  }'

Beberapa properti yang wajib diperhatikan:

• database.server.id harus unik dan tidak sama dengan server lain di cluster MySQL.
• topic.prefix menjadi awalan nama semua topic output.
• schema.history.internal.kafka.topic menyimpan riwayat skema agar MySQL tanpa binlog schema bisa tetap ter-rekonstruksi.

Verifikasi status connector dengan:

curl -s http://localhost:8083/connectors/inventory-connector/status

Output harus menunjukkan "state": "RUNNING" untuk connector dan task-nya. Jika masih FAILED, baca log container connect untuk melihat penyebabnya.

#Konfigurasi untuk Sumber Lain

Pola yang sama berlaku untuk database lain — hanya kelas connector dan properti koneksi yang berbeda.

#PostgreSQL

PostgreSQL memakai replication slot dan publication. Properti pentingnya antara lain plugin.name dengan nilai pgoutput, slot.name, dan publication.name:

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "database.hostname": "postgres",
  "database.port": "5432",
  "database.user": "postgres",
  "database.password": "postgres",
  "database.dbname": "postgres",
  "topic.prefix": "dbserver2",
  "schema.include.list": "public",
  "plugin.name": "pgoutput",
  "slot.name": "debezium_slot",
  "publication.name": "dbz_publication"
}

#MongoDB dan SQL Server

Untuk MongoDB, Debezium butuh connection string ke replica set dan topic.prefix. Untuk SQL Server, cukup host, port, nama database, dan skema yang ingin di-capture:

{
  "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
  "mongodb.connection.string": "mongodb://mongo:27017/?replicaSet=rs0",
  "mongodb.user": "debezium",
  "mongodb.password": "dbz",
  "topic.prefix": "mongo1"
}
{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.hostname": "sqlserver",
  "database.port": "1433",
  "database.user": "sa",
  "database.password": "Password!1",
  "database.names": "testdb",
  "schema.include.list": "dbo",
  "topic.prefix": "sqlserver1"
}

Catatan: MongoDB memerlukan mode replica set, bukan standalone, karena change streams hanya tersedia pada replica set.

#User Privileges yang Dibutuhkan

Privilege minimum mencegah connector menjadi lubang keamanan. Untuk MySQL, user Debezium membutuhkan hak SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, dan REPLICATION CLIENT:

Membuat user Debezium di MySQL

CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz';
GRANT SELECT, RELOAD, SHOW DATABASES,
      REPLICATION SLAVE, REPLICATION CLIENT
  ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

Untuk PostgreSQL, user harus punya hak REPLICATION dan SELECT pada tabel target. Jangan pernah memakai akun root untuk connector produksi — prinsip privilege minimal berlaku di semua database.

#Verifikasi Topic Output dan Pesan CDC Pertama

Setelah connector RUNNING, snapshot awal akan mengisi topic dengan event ber-op r. Untuk membuktikan streaming berjalan, tambahkan satu baris baru:

docker exec -it mysql bash -c 'mysql -u root -pdbz inventory -e "INSERT INTO customers VALUES (1001, \"Budi\", \"Santoso\", \"budi@example.com\")"'

Lalu baca topic dengan konsumen Kafka:

docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic dbserver1.inventory.customers \
  --from-beginning \
  --property print.key=true

Event yang tampil memiliki key berisi id baris dan value berisi objek payload dengan op bernilai c. Itu artinya instalasi berhasil: setiap perubahan pada tabel customers sekarang mengalir real-time ke Kafka.

#Penutup

Episode 3 membawa kalian dari nol ke pesan CDC pertama: menambahkan MySQL sebagai sumber, mendaftarkan connector lewat REST API, memahami konfigurasi untuk empat database populer, menyusun privilege minimum, dan memverifikasi event masuk ke topic.

Inti yang harus dibawa pulang:

• Connector didaftarkan lewat POST /connectors dengan body JSON.
• topic.prefix dan database.server.id adalah properti yang wajib dan harus unik.
• Setiap database punya mekanisme koneksi sendiri: pgoutput untuk PostgreSQL, replica set untuk MongoDB.
• Selalu buat user Debezium dengan privilege minimum.
• Verifikasi CDC dengan insert baris baru lalu baca topic dengan console consumer.

Di episode 4 selanjutnya kita akan membedah snapshot, streaming, dan offset management — bagaimana Debezium mengisi data awal, bagaimana offset disimpan dan dipulihkan, snapshot chunking untuk tabel besar, dan cara menangani perubahan skema.