Kafka Connect & Change Data Capture — Belajar Apache Kafka

Apa yang Dikerjakan Connect

Kafka Connect adalah kerangka kerja untuk memindahkan data masuk dan keluar Kafka tanpa menulis kode konsumen atau produser.

Source connector:  PostgreSQL / MySQL / MongoDB / S3  →  Kafka
Sink connector:    Kafka  →  Elasticsearch / S3 / BigQuery / JDBC

Yang Anda tulis hanyalah berkas konfigurasi JSON. Connect menangani distribusi tugas, penyimpanan offset, retry, dan pemulihan setelah kegagalan.

💡 Mode distributed, bahkan untuk satu node

Connect punya mode standalone dan distributed. Gunakan distributed sejak awal walau hanya menjalankan satu worker: konfigurasi dan offset disimpan di topik Kafka, sehingga konektor selamat dari restart dan bisa ditambah worker tanpa migrasi.

Change Data Capture dengan Debezium

CDC membaca log replikasi basis data — WAL pada PostgreSQL, binlog pada MySQL — dan mengubah tiap perubahan baris menjadi peristiwa Kafka.

UPDATE pesanan SET status='dibayar' WHERE id=123;
      │
      ▼  WAL PostgreSQL
  Debezium
      │
      ▼  topik: server.public.pesanan
  { "before": { "id":123, "status":"menunggu" },
    "after":  { "id":123, "status":"dibayar"  },
    "op": "u", "ts_ms": 1780000000000 }

⚠️ CDC bukan pengganti peristiwa domain

Peristiwa CDC menggambarkan perubahan baris tabel, bukan makna bisnis. Konsumen yang berlangganan CDC otomatis terikat pada skema internal basis data Anda — mengganti nama kolom akan merusak mereka. Untuk integrasi antar tim, terbitkan peristiwa domain yang dirancang sengaja; simpan CDC untuk replikasi data, gudang data, dan pola outbox.

🚫 Slot replikasi yang tertinggal bisa membuat basis data penuh

Pada PostgreSQL, konektor CDC memakai slot replikasi. Bila konektor mati atau tertinggal jauh, PostgreSQL menahan seluruh WAL yang belum dikonsumsi slot itu — dan disk basis data bisa penuh sampai instance berhenti melayani. Pantau pg_replication_slots, dan hapus slot yang konektornya sudah tidak dipakai lagi.

Contoh Konfigurasi Sink

{
  "name": "sink-elasticsearch-pesanan",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "pesanan",
    "connection.url": "http://elasticsearch:9200",
    "key.ignore": "false",
    "tasks.max": "3",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.sink-elasticsearch",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

⚠️ errors.tolerance=all tanpa dead letter queue = data hilang diam-diam

Setelan ini membuat Connect melewati pesan yang gagal alih-alih berhenti. Tanpa errors.deadletterqueue.topic.name, pesan yang gagal dibuang begitu saja dan Anda tidak akan pernah tahu berapa banyak yang hilang. Keduanya harus selalu dipasang bersama.

Single Message Transforms

SMT melakukan transformasi ringan pada tiap pesan tanpa perlu aplikasi terpisah: mengganti nama field, menutupi data sensitif, mengarahkan topik, menambah header.

"transforms": "samarkan,ubahTopik",
"transforms.samarkan.type": "org.apache.kafka.connect.transforms.MaskField$Value",
"transforms.samarkan.fields": "nomor_ktp,email",
"transforms.ubahTopik.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.ubahTopik.regex": "server.public.(.*)",
"transforms.ubahTopik.replacement": "cdc.$1"

💡 Batas wajar SMT

SMT bagus untuk transformasi per pesan yang sederhana. Begitu Anda membutuhkan penggabungan, agregasi, atau keadaan lintas pesan, pindah ke Kafka Streams atau Flink — memaksakan logika kompleks ke dalam rantai SMT menghasilkan konfigurasi yang tidak bisa diuji dan sulit ditelusuri saat gagal.