Apa Itu Kafka Connect? #

Dalam arsitektur sistem modern terdistribusi, Apache Kafka sering kali diposisikan sebagai sistem saraf pusat (central nervous system) yang mengalirkan data antar berbagai platform. Namun, memiliki broker Kafka yang cepat dan andal saja tidaklah cukup jika kita masih harus menulis kode kustom yang rumit setiap kali ingin memindahkan data dari database relasional ke Kafka, atau dari Kafka ke penyimpanan objek (object storage). Menghubungkan berbagai sistem eksternal secara manual sering kali berujung pada pembuatan aplikasi produser dan konsumen kustom yang rentan terhadap kegagalan, sulit diskalakan, dan memakan waktu pengembangan yang signifikan. Untuk memecahkan tantangan integrasi ini secara terstandarisasi, Apache Kafka menyediakan sebuah framework integrasi data yang sangat kuat bernama Kafka Connect. Artikel ini akan mengupas tuntas filosofi di balik Kafka Connect, mengapa kita membutuhkannya, arsitektur internalnya yang mencakup Connector, Task, dan Worker, serta mekanisme pengolahan data melalui Converter dan Single Message Transforms (SMT).

Mengapa Kita Membutuhkan Kafka Connect? #

Sebelum kita mendalami cara kerja internal Kafka Connect, mari kita tinjau terlebih dahulu mengapa penulisan kode integrasi manual menggunakan Kafka Producer API dan Consumer API standar sering kali menjadi mimpi buruk bagi tim pengembang (developer). Bayangkan kita memiliki skenario umum di mana kita harus menyalin setiap baris baru dari database transaksi PostgreSQL ke topik Kafka, lalu mengalirkan data tersebut dari Kafka ke Elasticsearch untuk kebutuhan pencarian cepat, serta ke AWS S3 untuk pengarsipan jangka panjang.

1. Kompleksitas Pengelolaan State dan Offset #

Jika kita menulis produser kustom untuk PostgreSQL, kita harus melacak baris mana saja yang sudah berhasil dikirim ke Kafka. Kita perlu menangani skenario jika aplikasi produser mati di tengah jalan: dari mana kita harus mulai membaca lagi saat aplikasi menyala kembali? Masalah ini semakin rumit di sisi konsumen (Elasticsearch dan S3). Konsumen harus secara konsisten melakukan commit offset ke Kafka hanya setelah data benar-benar tertulis di sistem target. Jika Elasticsearch mengalami penurunan performa (downtime), konsumen kita harus menahan pembacaan (backpressure) agar tidak terjadi kehilangan data.

2. Boilerplate Code yang Berulang #

Setiap kali ada database baru, sistem file baru, atau layanan cloud baru yang perlu dihubungkan dengan Kafka, tim pengembang terpaksa menulis kode boilerplate yang polanya serupa: melakukan koneksi ke sistem target, membaca data, melakukan serialisasi, mengirimkannya ke Kafka, menangani kegagalan jaringan, dan melakukan pemantauan (monitoring). Hal ini membuang waktu komputasi dan energi tim yang seharusnya bisa dialokasikan untuk menyelesaikan logika bisnis inti.

3. Masalah Skalabilitas dan Toleransi Kesalahan (Fault Tolerance) #

Menulis aplikasi integrasi satu instansi (single instance) mungkin terasa mudah di awal. Namun, bagaimana jika beban data tiba-tiba melonjak sepuluh kali lipat? Kita harus memikirkan cara membagi beban kerja pembacaan tabel database secara paralel lintas beberapa proses. Jika salah satu mesin yang menjalankan proses integrasi mati, bagaimana proses pemindahan tugas (failover) dilakukan ke mesin yang lain tanpa menyebabkan duplikasi data yang masif atau kehilangan pesan?

Kafka Connect dirancang untuk menyelesaikan semua tantangan di atas secara out-of-the-box. Dengan Kafka Connect, kita tidak lagi menulis kode pemrograman imperatif; kita cukup mendefinisikan konfigurasi deklaratif (biasanya dalam format JSON) untuk menjalankan pipa data (data pipeline) yang andal, toleran terhadap kegagalan, dan dapat diskalakan secara horizontal.

Filosofi Kafka Connect: Integrasi Tanpa Kode #

Filosofi inti dari Kafka Connect adalah data integration without coding. Ini berarti kita memisahkan logika integrasi data dari logika aplikasi bisnis kita. Kafka Connect menyediakan framework runtime terstandarisasi di mana siapa saja dapat menulis modul integrasi (connector) sekali, dan memanfaatkannya berulang kali di berbagai lingkungan runtime hanya dengan mengubah parameter konfigurasi.

Kapan Menggunakan Kafka Connect vs Kafka Streams? #

Sangat penting bagi kita untuk tidak mencampuradukkan peran Kafka Connect dengan Kafka Streams, karena keduanya memiliki tujuan arsitektural yang sepenuhnya berbeda:

Secara sederhana, jika kita ingin memindahkan data dari database PostgreSQL ke Kafka, gunakan Kafka Connect Source. Jika kita ingin mengubah data mentah transaksi di Kafka menjadi laporan agregat pendapatan per jam, gunakan Kafka Streams. Jika kita ingin memindahkan hasil agregat tersebut dari Kafka ke Elasticsearch, gunakan Kafka Connect Sink.

Tiga Pilar Utama Arsitektur Kafka Connect #

Arsitektur internal Kafka Connect didesain secara modular untuk memisahkan definisi logika integrasi, pembagian kerja paralel, dan penyediaan infrastruktur eksekusi. Tiga pilar utama yang menyusun arsitektur ini adalah Connector, Task, dan Worker.

flowchart TD
    subgraph WorkerCluster ["Kafka Connect Worker (Distributed Cluster)"]
        direction TB
        subgraph Worker1 ["Worker Instance 1"]
            C1["PostgreSQL Source Connector"]
            T1["Source Task 1"]
            T2["Source Task 2"]
        end
        subgraph Worker2 ["Worker Instance 2"]
            T3["Source Task 3"]
            T4["Sink Task 1"]
        end
    end
    
    SourceDB["(\"Database PostgreSQL\")"] -->|Read via JDBC/CDC| C1
    C1 -->|Membagi Pekerjaan| T1 & T2 & T3
    
    T1 & T2 & T3 -->|Kirim Event| KafkaCluster{"Apache Kafka Cluster"}
    
    KafkaCluster -->|Tarik Event| T4
    T4 -->|Write| TargetS3["AWS S3 Bucket"]

1. Connector #

Connector adalah komponen yang mendefinisikan logika interaksi dengan sistem eksternal tertentu. Ada dua jenis Connector:

• Source Connector: Bertugas menarik data dari sistem luar (seperti database relasional, antrean pesan lain, atau sensor IoT) dan mengirimkannya ke Apache Kafka sebagai record.
• Sink Connector: Bertugas mengambil data dari Apache Kafka dan menulisnya ke sistem target luar (seperti Elasticsearch, database NoSQL, atau penyimpanan objek cloud).

Connector tidak mengeksekusi proses pemindahan data secara langsung. Tugas utamanya adalah mendefinisikan konfigurasi, memeriksa sistem target (misalnya, mendeteksi tabel mana saja yang ada di database), dan menentukan bagaimana pekerjaan besar tersebut akan dibagi menjadi bagian-bagian kecil yang dapat dieksekusi secara paralel.

2. Task #

Task adalah unit kerja aktual dalam pemrosesan data di Kafka Connect. Connector membagi beban kerja yang besar menjadi satu atau beberapa Task. Sebagai contoh, sebuah PostgreSQL Source Connector dapat membagi pekerjaan pembacaan 10 tabel database menjadi 3 Task, di mana Task 1 membaca tabel 1-3, Task 2 membaca tabel 4-6, dan Task 3 membaca tabel 7-10.

Karena Task tidak menyimpan state operasional secara lokal (bersifat stateless), mereka dapat dijalankan di mana saja di dalam kluster. Task menerima data dari Connector (untuk Source) atau dari Kafka (untuk Sink) lalu memprosesnya. Skalabilitas horizontal di Kafka Connect dicapai secara langsung dengan meningkatkan jumlah Task yang diizinkan untuk berjalan secara paralel melalui konfigurasi tasks.max.

3. Worker #

Worker adalah proses runtime JVM aktual yang menjalankan Connector dan Task. Worker bertindak layaknya container eksekusi. Ada dua mode Worker yang dapat kita jalankan:

• Standalone Worker: Semua Connector dan Task dijalankan di dalam satu proses JVM tunggal di satu mesin. Mode ini sangat sederhana untuk dikonfigurasi namun tidak memiliki toleransi kegagalan dan kemampuan skalabilitas horizontal. Sangat cocok untuk pengembangan lokal (local development) atau ETL sederhana di tepi jaringan (edge node).
• Distributed Worker: Beberapa Worker berjalan secara terdistribusi pada beberapa mesin berbeda untuk membentuk sebuah kluster Connect. Mereka berkoordinasi secara otomatis menggunakan fitur internal Apache Kafka. Jika salah satu Worker mengalami crash, Worker lain yang tersisa akan mendeteksi hilangnya Worker tersebut dan secara otomatis memindahkan Task yang menggantung ke Worker yang sehat (auto-failover).

Model Data Internal: SourceRecord dan SinkRecord #

Salah satu kekuatan utama Kafka Connect adalah kemampuannya untuk mengintegrasikan sistem apa pun dengan sistem apa pun tanpa menciptakan ketergantungan erat (tight coupling). Hal ini dapat dicapai karena Kafka Connect mendefinisikan model data internal yang abstrak.

Ketika data ditarik dari PostgreSQL oleh JDBC Source Connector, data tersebut tidak langsung diubah menjadi byte JSON atau Avro. Sebaliknya, Connector mengubah representasi data database tersebut menjadi objek Java internal Kafka Connect yang bernama SourceRecord. Objek SourceRecord ini terdiri dari dua bagian utama:

1. Schema: Struktur data yang mendefinisikan nama-nama kolom, tipe data, dan apakah suatu kolom boleh bernilai null.
2. Value: Nilai aktual dari record tersebut yang dicocokkan dengan skema yang telah ditentukan.

Sebaliknya, pada sisi Sink, Kafka Connect akan mengambil byte data dari Kafka, mengubahnya menjadi objek internal bernama SinkRecord (yang juga memiliki Schema dan Value), lalu menyerahkannya ke Sink Connector untuk ditulis ke sistem luar seperti S3 atau Elasticsearch.

Dengan memisahkan format data internal dari format penyimpanan fisik di Kafka, satu connector yang sama dapat menulis data ke Kafka dalam format JSON, Avro, Protobuf, atau bahkan teks biasa tanpa perlu mengubah satu baris kode pun di dalam connector tersebut. Tugas penulisan format fisik diserahkan sepenuhnya kepada komponen yang disebut Converter.

Peran Converter dan Schema Registry #

Converter adalah komponen yang bertanggung jawab untuk menerjemahkan data antara model data internal Kafka Connect (skema dan nilai) dengan representasi biner ter-serialisasi yang disimpan di dalam topik Kafka. Converter dikonfigurasi secara independen dari Connector.

flowchart LR
    subgraph SourcePipeline ["Source Pipeline"]
        SourceSystem["Sistem Sumber"] -->|Data Mentah| SourceConn["Source Connector"]
        SourceConn -->|"SourceRecord (Internal)"| SrcConverter["Converter (e.g., Avro)"]
        SrcConverter -->|Kirim Skema| SchemaReg["Schema Registry"]
        SrcConverter -->|Byte Data Ter-serialisasi| KafkaTopic("(\"Kafka Topic\""))
    end
    
    subgraph SinkPipeline ["Sink Pipeline"]
        KafkaTopic -->|Byte Data Ter-serialisasi| SnkConverter["Converter (e.g., Avro)"]
        SchemaReg -.->|Unduh Skema| SnkConverter
        SnkConverter -->|"SinkRecord (Internal)"| SinkConn["Sink Connector"]
        SinkConn -->|Data Mentah| TargetSystem["Sistem Target"]
    end

Ada beberapa jenis Converter bawaan yang sering kita gunakan di lingkungan produksi:

1. StringConverter: Digunakan jika data kita berupa string teks sederhana (misalnya log mentah aplikasi).
2. JsonConverter: Mengubah model data internal menjadi string JSON biner. Kita dapat menyetel parameter schemas.enable=true untuk menyertakan skema data langsung di dalam setiap pesan JSON, meskipun ini akan menghasilkan overhead ukuran pesan yang sangat besar.
3. AvroConverter: Mengubah model data internal menjadi format biner Avro yang sangat ringkas. AvroConverter berintegrasi secara ketat dengan Confluent Schema Registry untuk menyimpan skema secara terpusat, sehingga record yang dikirim ke Kafka hanya berisi ID skema (5 byte) dan payload biner yang sangat kecil.
4. ProtobufConverter: Mirip dengan Avro, menggunakan format biner Protobuf dan memanfaatkan Schema Registry untuk manajemen kompatibilitas skema.

Mengonfigurasi Converter dalam Praktek #

Di bawah ini adalah contoh bagaimana kita mendefinisikan konfigurasi Converter pada berkas properti distributed worker (connect-distributed.properties):

# Menggunakan Avro Converter untuk kunci dan nilai pesan demi efisiensi biner
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

# ANTI-PATTERN: Jangan gunakan JsonConverter dengan schemas.enable=true di produksi skala besar
# karena akan menduplikasi metadata skema yang besar ke setiap record pesan.
# value.converter=org.apache.kafka.connect.json.JsonConverter
# value.converter.schemas.enable=true

Jika kita terpaksa menggunakan JSON tanpa Schema Registry namun tetap ingin mempertahankan informasi skema untuk sistem downstream (misalnya database SQL target), kita dapat mengaktifkan schemas.enable=true. Namun, untuk sistem produksi dengan throughput tinggi, sangat disarankan untuk beralih ke AvroConverter atau ProtobufConverter.

Single Message Transforms (SMT) #

Dalam perjalanan pipa data dari sistem sumber ke Kafka, atau dari Kafka ke sistem target, kita sering kali perlu melakukan modifikasi data tingkat sederhana. Contohnya, kita ingin menyembunyikan (masking) nomor kartu kredit pelanggan sebelum ditulis ke Kafka, menambahkan timestamp dinamis pada record, atau mengubah nama kolom agar sesuai dengan standar database target.

Untuk kebutuhan modifikasi data ringan baris-tunggal ini, Kafka Connect menyediakan fitur Single Message Transforms (SMT). SMT beroperasi langsung pada memori worker sebelum record di-serialisasikan oleh Converter (pada Source) atau tepat setelah record di-deserialisasikan oleh Converter (pada Sink).

Cara Kerja SMT #

SMT dapat dirangkai secara berurutan (chaining) untuk membentuk pipa transformasi data sederhana. Karena SMT beroperasi pada tingkat record tunggal (single record), ia tidak memiliki overhead penyimpanan state lokal dan memiliki latensi eksekusi yang sangat rendah (hitungan mikrodetik).

flowchart LR
    Source["Source Record"] --> T1["Transform 1: MaskField"] --> T2["Transform 2: InsertField"] --> Conv["Converter"] --> Byte["Kafka Byte"]

Beberapa modul SMT bawaan yang sangat berguna meliputi:

• MaskField: Mengganti nilai field tertentu dengan nilai statis atau null (sangat penting untuk kepatuhan regulasi privasi data seperti GDPR atau PCI-DSS).
• InsertField: Menyisipkan metadata tambahan seperti nama topik Kafka, timestamp sistem, atau partition ID ke dalam record.
• ReplaceField: Mengubah nama field atau menyaring field tertentu agar tidak ikut dikirim ke sistem target.
• Cast: Mengubah tipe data suatu field (misalnya mengubah string "123" menjadi integer 123).
• ValueToKey / RegExRouter: Mengubah struktur record atau memetakan pesan secara dinamis ke nama topik yang berbeda berdasarkan ekspresi reguler.

Contoh Konfigurasi SMT pada Source Connector #

Di bawah ini adalah contoh nyata bagaimana kita mengonfigurasi SMT di dalam file JSON konfigurasi Source Connector untuk menyembunyikan kolom kata sandi (password) dan menyisipkan UUID transaksi unik ke dalam record:

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://mysql-db:3306/shop",
    "connection.user": "exporter",
    "connection.password": "secret",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "mysql-",
    
    "//": "Mendefinisikan rantai transformasi bernama maskPassword dan addUUID",
    "transforms": "maskPassword,addUUID",
    
    "//": "Konfigurasi transformasi 1: Menyembunyikan isi kolom password",
    "transforms.maskPassword.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.maskPassword.fields": "password",
    "transforms.maskPassword.replacement": "[HIDDEN]",
    
    "//": "Konfigurasi transformasi 2: Menyisipkan nama topik Kafka ke dalam record",
    "transforms.addUUID.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addUUID.topic.field": "kafka_topic_origin"
  }
}

Batasan Kritis SMT #

Meskipun SMT sangat praktis, kita harus berhati-hati untuk tidak menyalahgunakannya. SMT tidak boleh digunakan untuk:

• Agregasi data lintas beberapa pesan (misal menghitung total penjualan per menit).
• Melakukan operasi join data dengan topik atau database lain secara dinamis.
• Melakukan kalkulasi logika bisnis yang kompleks dan memakan waktu I/O jaringan.

Jika pipa data kita membutuhkan operasi kompleks di atas, kita harus memindahkan logika tersebut ke layer pemrosesan aliran data aktual menggunakan Kafka Streams atau Apache Flink.

Ringkasan #

• Kafka Connect — Framework integrasi data deklaratif berbasis konfigurasi untuk memindahkan data masuk (Source) dan keluar (Sink) dari Apache Kafka tanpa menulis kode kustom.
• Connector vs Task — Connector bertugas mendefinisikan konfigurasi dan membagi beban kerja secara logis, sedangkan Task adalah unit eksekusi paralel yang berjalan di worker.
• Workers — Runtime engine Kafka Connect yang dapat berjalan dalam mode Standalone (untuk dev) atau Distributed (untuk produksi dengan skalabilitas tinggi dan toleransi kesalahan otomatis).
• Converters — Komponen penting yang menjembatani model data internal skema-nilai dengan representasi fisik (seperti Avro, Protobuf, atau JSON) saat berinteraksi dengan broker Kafka.
• Single Message Transforms (SMT) — Transformasi ringan baris-tunggal yang beroperasi di dalam memori JVM untuk melakukan pembersihan, manipulasi tipe, atau penyembunyian data sensitif sebelum serialisasi.

  ← Sebelumnya: EOS Limitation   Berikutnya: Source vs Sink Connector →