Belajar Apache Kafka - Kafka Connect: Integrasi Data

#Pendahuluan

Memindahkan data antara Kafka dan sistem eksternal — database, object storage, atau message queue — tidak perlu ditulis dari nol setiap kali. Kafka Connect adalah framework yang menjalankan connector: komponen yang menyalin data masuk (source) dari sistem lain ke Kafka, atau keluar (sink) dari Kafka ke sistem lain.

Nilai utama Kafka Connect adalah ia menangani hal-hal rumit yang biasanya menyita waktu: scaling dengan task paralel, pengelolaan offset, retry dan error handling, serta distribusi workload antar worker. Kalian cukup menulis file konfigurasi, dan framework mengurus sisanya.

Episode 12 ini akan membawa kalian memahami arsitektur worker, perbedaan source dan sink connector, REST API untuk mengelola connector, teknik deployment dengan task parallelism, dead letter queue, dan Single Message Transforms (SMT) untuk memodifikasi data di perjalanan.

#Arsitektur Kafka Connect

#Worker: Standalone vs Distributed

Connector dijalankan di dalam proses bernama worker. Dua mode yang didukung:

• Standalone: satu proses tunggal, cocok untuk development dan testing. Semua connector dan task berjalan di satu node.
• Distributed: banyak worker membentuk cluster, connector didistribusikan otomatis antar worker, mendukung failover dan scaling. Mode ini untuk produksi.

#Source dan Sink Connector

• Source connector: membaca dari sistem sumber (database, file, API) dan menulis ke Kafka. Contoh: JdbcSourceConnector membaca tabel database menjadi record.
• Sink connector: membaca dari Kafka dan menulis ke sistem tujuan. Contoh: S3SinkConnector menyimpan record menjadi file di object storage.

#Converter dan Transforms

Converter mengubah format data antara Kafka dan sistem eksternal: JsonConverter, AvroConverter, atau StringConverter. Transforms memodifikasi record di tengah perjalanan — misalnya menambah field atau mengubah key — sebelum mencapai sink. Urutan pemrosesan: converter source → transforms → writer, dan sebaliknya untuk sink.

#Connect REST API

Connect worker membuka REST API untuk mengelola connector:

curl -s http://localhost:8083/connectors

curl -s http://localhost:8083/connectors mengembalikan daftar nama connector dalam JSON. Endpoint lain: GET /connectors/{name} untuk status, PUT /connectors/{name}/config untuk update, dan DELETE /connectors/{name} untuk menghapus.

#Built-in dan Community Connectors

#File dan JDBC Connectors

Distribusi Kafka menyertakan dua connector dasar: FileStreamSource dan FileStreamSink. Untuk integrasi database, JdbcSourceConnector membaca tabel atau query menjadi topic, dan JdbcSinkConnector menulis record ke tabel:

{
  "name": "jdbc-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://localhost:5432/app",
    "auto.create": "true",
    "insert.mode": "upsert"
  }
}

connector.class menentukan implementasi yang dipakai, dan connection.url menunjuk database tujuan. Ubah nama connector untuk menambahkan lebih banyak instance.

#Community dan Connector Hub

Ekosistem Kafka Connect sangat kaya: connector untuk MongoDB, Elasticsearch, MySQL, PostgreSQL, AWS S3, Azure Blob, dan ratusan lainnya tersedia di Connector Hub Confluent. Sebelum menulis connector sendiri, cek dahulu apakah connector yang dibutuhkan sudah tersedia — menulis connector custom adalah pekerjaan terakhir yang kalian inginkan.

#Deployment dan Manajemen Connector

#Task Parallelism

Satu connector membagi kerja menjadi beberapa task yang berjalan paralel. tasks.max menentukan jumlah maksimum task:

{
  "name": "s3-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders",
    "s3.bucket.name": "data-orders",
    "flush.size": "10000",
    "tasks.max": "4"
  }
}

tasks.max=4 memungkinkan sink membaca 4 partition berbeda secara paralel. Aturan penting: jumlah task sink tidak boleh melebihi jumlah partition topic, karena satu partition hanya bisa dibaca satu task dalam satu group.

#Error Handling dan Dead Letter Queue

Connector yang gagal memproses record bisa dikonfigurasi untuk mengirim record bermasalah ke dead letter queue (DLQ) alih-alih menghentikan pipeline:

{
  "name": "jdbc-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "orders-errors",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

Dengan errors.tolerance=all, connector melanjutkan pemrosesan record lain dan menulis record yang gagal ke topic orders-errors beserta header konteks error — informasi yang sangat berharga untuk debugging.

#Single Message Transforms (SMT)

#Transformasi Umum

SMT memodifikasi record satu per satu sebelum ditulis ke sink:

• InsertField: menambah field statis, misalnya timestamp pipeline.
• ReplaceField: menghapus atau mengganti nama field.
• Flatten: meratakan struktur nested menjadi flat.
• Cast: mengubah tipe data field, misalnya string ke integer.
• TimestampRouter: menulis record ke topic berbeda berdasarkan timestamp, berguna untuk log berpartisi waktu.

#Contoh Konfigurasi SMT

{
  "name": "s3-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders",
    "transforms": "insertTs,router",
    "transforms.insertTs.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.insertTs.timestamp.field": "ingested_at",
    "transforms.router.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "transforms.router.topic.format": "orders-${timestamp}"
  }
}

Rantai transformasi didefinisikan dengan alias yang dipisahkan koma pada transforms. SMT dijalankan berurutan: insertTs menambah field ingested_at, lalu router mengarahkan record ke topic orders-YYYY-MM-DD berdasarkan timestamp — pola umum untuk log berpartisi waktu.

#Penutup

Di episode 12 ini kalian sudah memahami arsitektur Kafka Connect, perbedaan worker standalone dan distributed, source dan sink connector, converter dan transforms, REST API untuk manajemen, task parallelism, dead letter queue, serta Single Message Transforms.

Inti yang harus dibawa pulang:

• Source connector menulis ke Kafka; sink connector membaca dari Kafka.
• Distributed mode untuk produksi: scaling dan failover otomatis antar worker.
• REST API di port 8083 mengelola seluruh connector.
• tasks.max menentukan paralelisme; sink task tidak boleh melebihi jumlah partition.
• DLQ dengan errors.tolerance=all mencegah satu record rusak menghentikan pipeline.
• SMT memodifikasi record di perjalanan tanpa menulis kode.

Di episode 13 selanjutnya kita akan membahas Kafka Streams — library stream processing untuk memfilter, mengagregasi, dan menggabungkan data yang mengalir. Kalian akan belajar KStream, KTable, state store dengan RocksDB, windowing, serta processing guarantee exactly-once.