Kafka Connect

Kafka Connect adalah kerangka integrasi bawaan Kafka yang memindahkan data antara Kafka dan sistem lain secara andal, misalnya dari basis data ke topic untuk diproses lebih lanjut atau dari topic ke penyimpanan objek sebagai arsip. Topik ini membahas model connector source dan sink, perbedaan dua mode worker beserta konfigurasi worker dan connector, serta REST API pada port 8083 yang melayani pengelolaan connector. Setelah membaca topik ini, pemilihan mode worker dan susunan konfigurasi untuk sebuah integrasi dapat ditentukan dengan dasar yang jelas.

Model connector source dan sink#

Kafka Connect membagi pekerjaan penyalinan data menjadi dua lapis. Connector adalah komponen yang menerima konfigurasi dan memutuskan cara pekerjaan dibagi, sedangkan task adalah unit paralelisme yang benar-benar menyalin data. Connector membaca konfigurasi seperti tasks.max, lalu menghasilkan satu atau beberapa task dengan pembagian pekerjaan miliknya, misalnya satu task untuk satu tabel. Task yang dihasilkan bisa lebih sedikit dari batas tasks.max apabila sumber data tidak menyediakan pekerjaan sebanyak itu.

Arah pindahnya data membedakan dua jenis connector. Source connector membaca data dari sistem eksternal dan menulisnya ke topic Kafka, sedangkan sink connector mengonsumsi pesan dari topic dan menulisnya ke sistem eksternal. Data yang dipindahkan berbentuk record Connect, yaitu pasangan key dan value dengan struktur yang dikenal Connect. Record diterjemahkan menjadi byte oleh converter, yaitu komponen yang menerjemahkan struktur data Connect ke format tersimpan seperti JSON atau Avro. Posisi pekerjaan dikelola otomatis oleh kerangka Connect, yaitu source connector menyimpan posisi baca di sistem sumber sebagai offset, sedangkan sink connector menyimpan offset Kafka yang sudah diproses. Setelah proses mati dan dijalankan ulang, pekerjaan dilanjutkan dari posisi terakhir tanpa kode tambahan.

Alur data dari sistem sumber ke sistem tujuan#

Alur lengkapnya dapat digambar sebagai empat langkah. Connector source membaca data dari sistem sumber, mengubahnya menjadi record, lalu menulisnya ke topic. Di sisi lain, connector sink mengonsumsi pesan topic yang sama dan menulisnya ke sistem tujuan. Diagram berikut merangkum keempat langkah tersebut.

Demo berikut menyimulasikan alur itu dengan Python murni tanpa broker dan tanpa connector sungguhan. Connector source meniru FileStreamSourceConnector bawaan yang membaca berkas baris demi baris, sedangkan connector sink meniru FileStreamSinkConnector yang menulis pesan ke berkas lain. Topic disimulasikan sebagai list pesan per partisi, dan converter JsonConverter disimulasikan membungkus setiap nilai dengan envelope skema sebagaimana bentuk keluaran JsonConverter sungguhan ketika value.converter.schemas.enable bernilai true.

# Simulasi model Kafka Connect dengan Python murni: connector source membaca
# berkas baris demi baris dan menulisnya ke topic, lalu connector sink membaca
# topic dan menulisnya ke berkas tujuan. Broker tidak dilibatkan, sehingga
# topic disimulasikan sebagai list pesan per partisi.

import json

berkas_sumber = ["suhu=26.1", "suhu=26.4", "suhu=26.7"]
JUMLAH_PARTISI = 2

# --- connector source: baca berkas sumber, tulis pesan ke topic ---
print("=== connector source: sensor.txt -> topic ===")
topic = {}   # partisi -> list pesan terserialisasi
for offset_sumber, nilai in enumerate(berkas_sumber):
    # converter worker (JsonConverter dengan schemas.enable=true) membungkus nilai
    pesan = json.dumps({"schema": {"type": "string", "optional": False}, "payload": nilai})
    partisi = offset_sumber % JUMLAH_PARTISI   # pembagian partisi disederhanakan
    topic.setdefault(partisi, []).append(pesan)
    print(f"offset sumber={offset_sumber} nilai='{nilai}' -> partisi={partisi} offset={len(topic[partisi]) - 1}")
    print(f"  pesan di topic: {pesan}")

# --- connector sink: baca pesan dari topic, tulis ke berkas tujuan ---
print("=== connector sink: topic -> sensor-terkirim.txt ===")
berkas_tujuan = []
for partisi in sorted(topic):
    for offset, pesan in enumerate(topic[partisi]):
        nilai = json.loads(pesan)["payload"]
        berkas_tujuan.append(nilai)
        print(f"partisi={partisi} offset={offset} nilai='{nilai}' -> tulis baris {len(berkas_tujuan)}")

print(f"berkas tujuan: {berkas_tujuan}")

=== connector source: sensor.txt -> topic ===
offset sumber=0 nilai='suhu=26.1' -> partisi=0 offset=0
  pesan di topic: {"schema": {"type": "string", "optional": false}, "payload": "suhu=26.1"}
offset sumber=1 nilai='suhu=26.4' -> partisi=1 offset=0
  pesan di topic: {"schema": {"type": "string", "optional": false}, "payload": "suhu=26.4"}
offset sumber=2 nilai='suhu=26.7' -> partisi=0 offset=1
  pesan di topic: {"schema": {"type": "string", "optional": false}, "payload": "suhu=26.7"}
=== connector sink: topic -> sensor-terkirim.txt ===
partisi=0 offset=0 nilai='suhu=26.1' -> tulis baris 1
partisi=0 offset=1 nilai='suhu=26.7' -> tulis baris 2
partisi=1 offset=0 nilai='suhu=26.4' -> tulis baris 3
berkas tujuan: ['suhu=26.1', 'suhu=26.7', 'suhu=26.4']

Keluaran memperlihatkan tiga hal yang menentukan model Connect. Baris offset sumber menyatakan posisi baca di berkas sumber, yaitu offset yang disimpan source connector agar pembacaan dapat dilanjutkan setelah proses berhenti. Baris pesan di topic memperlihatkan bentuk pesan setelah converter bekerja, yaitu nilai asli dibungkus envelope schema dan payload. Baris terakhir menegaskan bahwa urutan berkas tujuan menjadi 26.1, 26.7, lalu 26.4, karena Kafka hanya menjamin urutan di dalam satu partisi dan sink membaca partisi satu per satu. Pada demo ini pembagian partisi disederhanakan, tetapi sifat urutan per partisinya berlaku sama pada Connect sungguhan.

Worker standalone dan distributed#

Connector dan task tidak berjalan sendiri, melainkan di dalam worker, yaitu proses JVM yang menyatukan kerangka Connect, converter, dan plugin connector. Worker tersedia dalam dua mode. Mode standalone menjalankan seluruh pekerjaan dalam satu proses, dengan konfigurasi connector diberikan sebagai berkas pada baris perintah dan offset disimpan di berkas lokal melalui offset.storage.file.filename. Mode ini sederhana untuk pengembangan dan kasus satu mesin, tetapi tidak memiliki ketahanan, karena worker yang mati menghentikan seluruh pekerjaan.

Mode distributed menjalankan beberapa worker yang membentuk satu cluster dengan identitas group.id dan berkoordinasi melalui protokol group yang sama dengan consumer group. Konfigurasi connector, offset, dan status task tidak disimpan di berkas, melainkan di tiga topic Kafka yang dikompaksi, yaitu config.storage.topic, offset.storage.topic, dan status.storage.topic. Worker yang baru bergabung langsung menerima pembagian task, sedangkan worker yang mati menyebabkan task-nya dipindahkan ke worker lain tanpa perubahan konfigurasi. Perintah berikut menjalankan worker standalone dengan satu berkas konfigurasi connector, dan bentuk yang sama digunakan untuk mode distributed melalui connect-distributed.sh.

# tidak-dijalankan
# masuk ke folder distribusi Kafka lalu jalankan worker beserta berkas konfigurasinya
cd /home/kal/my-project/kafka-runtime/kafka
bin/connect-standalone.sh config/connect-standalone.properties config/connector-file-source.json

Perbedaan keduanya dirangkum dalam tabel berikut.

Konfigurasi worker dan connector#

Konfigurasi worker berupa berkas properties yang berlaku untuk seluruh connector di dalam proses itu. Semua mode wajib menetapkan bootstrap.servers, key.converter, dan value.converter, sedangkan mode standalone menambahkan offset.storage.file.filename. Converter menentukan format pesan yang ditulis ke dan dibaca dari Kafka, dan pilihan JSON atau Avro di sini membuat connector apa pun dapat bekerja dengan format apa pun tanpa mengubah kode connector. Berkas berikut adalah contoh konfigurasi worker standalone yang mencakup kunci-kunci utamanya.

# konfigurasi worker Connect dalam mode standalone
bootstrap.servers=localhost:9092

# converter menentukan bentuk pesan yang ditulis ke dan dibaca dari Kafka
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

# pada mode standalone, offset connector source disimpan di berkas lokal
offset.storage.file.filename=/tmp/m6-connect.offsets
offset.flush.interval.ms=10000

# daftar lokasi plugin connector (connector, converter, transformation)
plugin.path=/opt/kafka/plugins

Konfigurasi connector berupa pasangan kunci dan nilai yang menetapkan connector mana yang dijalankan beserta perilakunya. Kunci yang umum adalah name sebagai pengenal tunggal, connector.class sebagai nama kelas Java, tasks.max sebagai batas task, serta key.converter dan value.converter untuk menimpa converter milik worker. Sink connector wajib menetapkan topics atau topics.regex sebagai sumber pesannya, sedangkan kunci lainnya bergantung pada connector yang dipilih. Berkas JSON berikut mendaftarkan connector source bawaan yang membaca berkas sensor.txt dan menulisnya ke topic sensor-suhu, dan bentuk yang sama digunakan sebagai body permintaan POST /connectors pada mode distributed.

{
  "name": "source-sensor-berkas",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "sensor.txt",
    "topic": "sensor-suhu"
  }
}

Nilai tasks.max di atas bernilai satu karena satu berkas tidak memberi pekerjaan untuk lebih dari satu task. connector.class juga boleh ditulis pendek sebagai FileStreamSource, karena kerangka Connect menerima alias nama kelas tanpa paket maupun akhiran Connector. Nilai name menjadi pengenal tunggal connector di dalam cluster, sehingga pendaftaran dengan nama yang sudah digunakan connector lain akan ditolak.

REST API pada port 8083#

Worker melayani REST API di alamat bawaan http://localhost:8083, dan API ini tersedia pada kedua mode sehingga connector juga dapat didaftarkan saat worker standalone sudah berjalan. Pada mode distributed, API ini sekaligus menjadi saluran komunikasi antar worker, yaitu permintaan yang diterima worker follower diteruskan ke worker leader. Endpoint yang paling sering digunakan adalah GET /connectors untuk daftar connector aktif, POST /connectors untuk pendaftaran dengan body berisi name dan config, GET /connectors/{name}/status untuk memantau keadaan, PUT /connectors/{name}/config untuk mengubah konfigurasi, dan DELETE /connectors/{name} untuk menghapus connector.

Dua perintah berikut mendaftarkan connector dari berkas JSON yang sudah dibuat sebelumnya, kemudian membaca statusnya. Keduanya memerlukan worker Connect yang berjalan, sehingga hanya bentuk permintaannya yang ditampilkan.

# tidak-dijalankan
# daftarkan connector dengan body dari berkas JSON yang sudah dibuat
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @connector-file-source.json

# tidak-dijalankan
# baca keadaan connector beserta keadaan setiap task-nya
curl http://localhost:8083/connectors/source-sensor-berkas/status

Respons status berisi keadaan connector beserta keadaan setiap task-nya, yaitu RUNNING ketika normal, FAILED ketika terjadi error beserta pesannya, serta worker tempat task itu berjalan. Endpoint GET /connector-plugins juga tersedia untuk memeriksa daftar connector yang terpasang pada worker yang menangani permintaan, sehingga keberhasilan pemasangan plugin dapat diverifikasi sebelum connector didaftarkan. Pemantauan dan pengubahan lewat endpoint ini tidak menghentikan worker, sehingga aliran data yang sedang berjalan tidak terputus.

Kesalahan umum#

Tiga kegagalan berikut paling sering muncul ketika connector pertama didaftarkan. Setiap kegagalan dilengkapi penyebab dan perbaikannya, sehingga diagnosisnya dapat dilakukan berurutan dari pemasangan plugin, konfigurasi converter, hingga cara pendaftaran connector.

Worker menampilkan error berupa kelas connector yang tidak ditemukan saat connector didaftarkan. Penyebabnya adalah jar connector tidak berada di plugin.path atau dependensinya terpisah dari jar utama. Perbaikannya berupa penempatan jar beserta seluruh dependensinya dalam satu direktori di bawah plugin.path, kemudian worker dijalankan ulang dan verifikasi dilakukan lewat GET /connector-plugins. Keluaran endpoint itu, yang memuat nama kelas connector, menandai plugin sudah terbaca oleh worker.

Consumer gagal mengurai pesan yang ditulis connector. Penyebab yang lazim adalah nilai true pada key.converter.schemas.enable dan value.converter.schemas.enable, sehingga setiap pesan terbungkus envelope schema dan payload, sedangkan consumer mengharapkan JSON polos. Perbaikannya berupa penyamaan konfigurasi converter di kedua sisi, yaitu mematikan envelope tersebut atau menggunakan deserializer yang memahaminya.

Perubahan berkas konfigurasi connector tidak berefek pada mode distributed. Konfigurasi connector tersimpan di config.storage.topic sejak pendaftarannya melalui REST API, sehingga penyuntingan berkas lokal tidak pernah dibaca ulang oleh worker. Perbaikannya berupa pengubahan konfigurasi melalui PUT /connectors/{name}/config atau pendaftaran ulang dengan POST /connectors.

Ringkasan#

• Kafka Connect memindahkan data melalui connector source dan sink yang pekerjaannya dijalankan task di dalam worker, dengan offset dikelola otomatis oleh kerangka Connect.
• Mode standalone menjalankan satu proses dengan konfigurasi connector dari berkas dan offset di berkas lokal, sedangkan mode distributed membentuk cluster worker yang menyimpan konfigurasi, offset, dan status di topic Kafka terkompaksi.
• Konfigurasi worker menetapkan alamat broker, converter, dan plugin.path, sedangkan konfigurasi connector menetapkan name, connector.class, tasks.max, dan kunci khusus connector.
• REST API pada port 8083 mendaftarkan, memantau, dan mengubah connector tanpa menghentikan worker.

Referensi#

• Kafka Connect Overview — Apache Kafka 4.4 Documentation — peran Kafka Connect, daftar fiturnya, dan dua mode eksekusi worker.
• Kafka Connect User Guide — Apache Kafka 4.4 Documentation — konfigurasi worker dan connector, topic penyimpanan mode distributed, plugin.path, serta daftar endpoint REST API.
• Kafka Connect — Confluent Documentation — dokumentasi ekosistem connector dan konsep Connect pada Confluent Platform.