Connector umum

Tiga tujuan integrasi yang paling sering ditemui pada pekerjaan sehari-hari adalah basis data relasional, penyimpanan objek, dan mesin pencarian, sehingga connector untuk ketiganya menjadi yang paling kerap dipasang pada worker Connect. Topik ini menampilkan profil connector JDBC, connector S3, dan connector Elasticsearch, mulai dari arah alir data, kebutuhan yang menyertainya, hingga kunci konfigurasinya. Pembahasan dibuka dengan cara pemasangan connector melalui plugin.path dan ditutup dengan simulasi sink JDBC sederhana untuk melihat cara pesan dipetakan menjadi baris tabel.

Instalasi connector melalui plugin.path#

Connector tidak berada di dalam broker, melainkan dimuat oleh worker Connect sebagai plugin, yaitu jar connector beserta seluruh dependensinya. Distribusi Kafka hanya membawa contoh FileStream dalam connect-file-4.4.0.jar yang tetap tidak dimuat sebelum plugin.path menunjuk ke lokasinya. Connector JDBC, S3, dan Elasticsearch pun dipasang sebagai paket terpisah yang tersedia melalui Confluent Hub, lalu letaknya didaftarkan pada kunci plugin.path milik worker. Kunci tersebut berupa daftar jalur yang dipisahkan koma, dan setiap jalurnya dipindai hingga ke subdirektori di dalamnya untuk menemukan jar connector, converter, maupun transformation.

Bentuk pemasangan yang lazim adalah satu subdirektori untuk satu connector, yaitu jar utama dan semua dependensinya berada dalam folder yang sama. Worker memberi setiap subdirektori classloader yang terisolasi, sehingga dua connector yang membutuhkan versi pustaka berbeda tidak saling menimpa. Setelah worker dijalankan ulang, pemasangan diverifikasi dengan GET /connector-plugins sebelum connector didaftarkan. Pada cluster multi-node, jar connector beserta dependensinya wajib terpasang di setiap worker, karena task dapat ditempatkan di worker mana pun.

Connector JDBC#

Connector JDBC menyediakan dua arah untuk basis data relasional. Sink mengekspor pesan topic ke tabel dengan menjalankan pernyataan SQL melalui driver JDBC, sedangkan source menyalin isi tabel ke topic dengan menjalankan query secara berkala. Kedua arah menggunakan connection.url sebagai alamat basis data, dan driver basis data yang dituju harus tersedia di worker karena paket connector tidak menyertakan driver untuk semua basis data.

Sink JDBC memetakan nama topic ke nama tabel melalui table.name.format, lalu menulisnya dengan insert.mode bernilai insert atau upsert. Sumber primary key ditentukan oleh pk.mode, yaitu none, record_key, record_value, atau koordinat Kafka, sedangkan auto.create dan auto.evolve membuat tabel yang belum ada serta menambah kolom baru dari skema record. Perlu dicatat bahwa pesan tombstone membuat task gagal secara bawaan, dan penghapusan baris hanya berjalan setelah delete.enabled=true dengan pk.mode=record_key.

Source JDBC menentukan cara mendeteksi data baru melalui mode. Nilai bulk menyalin seluruh tabel pada setiap iterasi, incrementing bergantung pada kolom yang nilainya terus naik, timestamp bergantung pada kolom waktu perubahan, sedangkan timestamp+incrementing menggabungkan keduanya dan menjadi mode yang paling kokoh. Frekuensi query diatur melalui poll.interval.ms. Pendekatan polling ini selalu menyisakan jeda, sehingga Debezium menjadi alternatif ketika jeda tidak dapat diterima. Debezium adalah keluarga connector change data capture yang membaca log transaksi basis data, sehingga perubahan, termasuk penghapusan baris, terdeteksi hampir seketika.

Berkas berikut mengonfigurasi sink JDBC yang mengonsumsi topic stok-gudang dan menulisnya ke basis data gudang dengan upsert per key dan penghapusan baris untuk tombstone.

{
  "name": "sink-jdbc-stok",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "stok-gudang",
    "connection.url": "jdbc:postgresql://localhost:5432/gudang",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "delete.enabled": "true",
    "auto.create": "true"
  }
}

Connector S3#

Connector S3 adalah sink yang menulis record topic ke objek di bucket S3. Record dari setiap partisi Kafka dipecah menjadi chunk, yaitu kumpulan record yang ditulis menjadi satu objek, dan nama objeknya memuat topic, nomor partisi, serta offset awal chunk tersebut. Bentuk isi objek dipilih melalui format.class dengan pilihan Avro, JSON, Parquet, atau Bytes, sehingga keluarannya dapat langsung diolah oleh mesin analisis yang mendukung format tersebut.

Penulisan objek ditutup oleh pemicu rotasi. flush.size menetapkan jumlah record maksimum per objek tanpa nilai bawaan, rotate.interval.ms menetapkan waktu maksimum sebuah objek tetap terbuka, sedangkan rotate.schedule.interval.ms menutup objek pada batas waktu yang dijadwalkan. Ketiga pemicu ini dapat digabungkan, dan rotasi terjadi setiap kali salah satu pemicu terpenuhi. Kredensial bucket diatur melalui variabel lingkungan atau berkas kredensial pada worker, dengan s3.bucket.name dan s3.region menentukan bucket tujuannya.

Penataan jalur objek dilakukan oleh partitioner. Partitioner bawaan mempertahankan partisi Kafka dalam nama objek, sedangkan TimeBasedPartitioner menata objek menurut waktu melalui path.format, misalnya 'year'=YYYY/'month'=MM/'day'=dd, dengan sumber waktu ditentukan oleh timestamp.extractor bernilai Wallclock, Record, atau RecordField. Partitioner yang deterministik memberi jaminan exactly-once bagi pembaca objek, karena nama objek yang sama selalu berisi chunk yang sama. Susunan inilah yang menjadikan sink S3 pilihan utama untuk arsip analitik.

Connector Elasticsearch#

Connector Elasticsearch adalah sink yang menulis record ke cluster Elasticsearch, dengan connection.url menunjuk alamat cluster, misalnya http://localhost:9200. Nama topic digunakan langsung sebagai nama index tujuan secara bawaan, sehingga topic stok-gudang diindeks ke index stok-gudang tanpa konfigurasi tambahan. Connector versi terkini mendukung Elasticsearch 8.x dan 9.x, sehingga versi cluster menjadi pertimbangan pertama sebelum pemasangan.

Id dokumen ditentukan secara deterministik, yaitu dari key record, atau berupa topic+partition+offset ketika key.ignore=true. Id yang deterministik membuat penulisan bersifat idempoten, karena record yang terkirim ulang menimpa dokumen dengan id yang sama dan tidak menduplikasinya. schema.ignore=false membuat mapping disusun dari skema record pertama dan tidak diperbarui setelahnya, sedangkan behavior.on.null.values menentukan perlakuan terhadap tombstone dan write.method=upsert mengaktifkan pembaharuan dokumen. Kunci-kunci ini menjaga bentuk data agar sama persis di Kafka dan di index Elasticsearch.

Simulasi sink JDBC#

Demo berikut menyimulasikan cara sink JDBC memetakan pesan menjadi baris tabel dengan Python murni. Konfigurasi yang ditiru sama dengan berkas JSON pada bagian JDBC, yaitu insert.mode=upsert, pk.mode=record_key, dan delete.enabled=true. Topic disimulasikan sebagai lima pesan beroffset berurutan, dan dua pesan di antaranya menyasar key yang sama. Offset hanya menjadi label urutan pada simulasi ini, sedangkan keputusan pemetaannya berlaku sama pada pesan yang dibaca sink JDBC sungguhan.

# Simulasi sink JDBC sederhana dengan Python murni: setiap pesan topic diubah
# menjadi baris tabel. Semantiknya meniru insert.mode=upsert dengan
# pk.mode=record_key, ditambah pesan tombstone (key terisi, nilai null) yang
# menghapus baris dengan key yang sama sebagaimana delete.enabled=true.

import json

# pesan topic stok-gudang: (offset, key, nilai JSON), nilai null berarti tombstone
pesan_topic = [
    (0, "P-101", '{"produk": "Pulpen", "jumlah": 30}'),
    (1, "P-102", '{"produk": "Buku Tulis", "jumlah": 10}'),
    (2, "P-103", '{"produk": "Penghapus", "jumlah": 5}'),
    (3, "P-101", '{"produk": "Pulpen", "jumlah": 24}'),
    (4, "P-102", None),
]

tabel = {}   # primary key -> baris (dict kolom: nilai)
for offset, key, nilai in pesan_topic:
    if nilai is None:
        tabel.pop(key, None)
        print(f"offset={offset} key={key} nilai=null -> delete baris {key}")
    else:
        baris = json.loads(nilai)
        aksi = "update" if key in tabel else "insert"
        tabel[key] = baris
        print(f"offset={offset} key={key} nilai='{nilai}' -> {aksi} baris {key}")

print(f"== isi tabel stok ({len(tabel)} baris) ==")
for key in sorted(tabel):
    print(f"{key} : {tabel[key]}")

offset=0 key=P-101 nilai='{"produk": "Pulpen", "jumlah": 30}' -> insert baris P-101
offset=1 key=P-102 nilai='{"produk": "Buku Tulis", "jumlah": 10}' -> insert baris P-102
offset=2 key=P-103 nilai='{"produk": "Penghapus", "jumlah": 5}' -> insert baris P-103
offset=3 key=P-101 nilai='{"produk": "Pulpen", "jumlah": 24}' -> update baris P-101
offset=4 key=P-102 nilai=null -> delete baris P-102
== isi tabel stok (2 baris) ==
P-101 : {'produk': 'Pulpen', 'jumlah': 24}
P-103 : {'produk': 'Penghapus', 'jumlah': 5}

Keluaran memperlihatkan tiga keputusan pemetaan. Tiga pesan pertama menyisipkan baris baru karena key-nya belum ada pada tabel, sedangkan pesan offset 3 membawa key yang sudah ada sehingga tindakannya menjadi update sebagaimana semantik upsert. Pesan terakhir berupa tombstone, yaitu pesan dengan key terisi dan nilai kosong, sehingga baris P-102 dihapus dan tabel menyisakan dua baris. Baris P-102 hilang meskipun pesannya pernah disisipkan pada offset 1, karena keputusan terakhir untuk satu key selalu berasal dari pesan dengan offset terbesar. Pada connector JDBC sungguhan, keputusan-keputusan itu menjadi pernyataan INSERT, UPDATE, dan DELETE yang dikirim ke basis data, dengan urutan pemrosesan tetap mengikuti urutan offset.

Kesalahan umum#

Tiga kegagalan berikut muncul paling sering saat connector JDBC, S3, dan Elasticsearch dipasang pertama kali. Masing-masing dilengkapi penyebab dan perbaikannya, sehingga diagnosisnya bergerak dari isi direktori plugin, kunci konfigurasi, hingga bentuk data yang dikirim.

Kegagalan pertama yang lazim adalah driver JDBC yang hilang. Connector berhasil didaftarkan, tetapi task gagal dengan error berupa kelas driver yang tidak ditemukan, karena paket connector tidak menyertakan driver untuk setiap basis data. Perbaikannya berupa penempatan jar driver ke dalam direktori plugin yang sama pada setiap worker, lalu task dijalankan ulang.

Connector S3 gagal dijalankan ketika rotate.schedule.interval.ms diisi tanpa timezone. Connector menghentikan pekerjaannya dengan pesan error saat mulai berjalan, karena zona waktu wajib ditetapkan lebih dahulu untuk rotasi terjadwal. Perbaikannya berupa pengisian timezone setiap kali rotate.schedule.interval.ms digunakan.

Index Elasticsearch menolak record setelah skema berubah. Connector hanya menyusun mapping dari skema record pertama dan tidak memperbaruinya, sehingga record dengan tipe field yang berbeda ditolak dan berakhir di dead letter queue. Perbaikannya berupa perancangan mapping index sejak awal dan pemantauan pesan yang terkumpul di DLQ. Record yang ditolak tidak menghentikan connector, tetapi kemajuan offsetnya perlu dipantau agar data tidak luput.

Ringkasan#

• Connector JDBC menghubungkan Kafka dengan basis data relasional pada dua arah, dengan mode bulk, incrementing, timestamp, dan timestamp+incrementing pada source serta insert.mode, pk.mode, dan auto.create pada sink.
• Connector S3 menulis objek per chunk dalam format Avro, JSON, Parquet, atau Bytes, dengan rotasi diatur oleh flush.size dan interval waktu serta jalur objek ditata oleh partitioner.
• Connector Elasticsearch menulis dokumen dengan id deterministik dari key atau topic+partition+offset, sehingga pengiriman ulang tidak menduplikasi dokumen.
• Ketiga connector dipasang sebagai plugin melalui plugin.path, satu subdirektori per connector, dengan driver basis data menyertai connector JDBC.

Referensi#

• JDBC Sink Connector for Confluent Platform — Confluent Documentation — pemetaan data, key handling, delete mode, dan auto-creation pada sink JDBC.
• JDBC Source Connector for Confluent Platform — Confluent Documentation — mode bulk, incrementing, timestamp, dan query berkala pada source JDBC.
• Amazon S3 Sink Connector for Confluent Platform — Confluent Documentation — format objek, strategi rotasi, dan partitioner pada sink S3.
• Elasticsearch Service Sink Connector for Confluent Platform — Confluent Documentation — penulisan dokumen, id deterministik, dan versi Elasticsearch yang didukung.
• Kafka Connect User Guide — Apache Kafka 4.4 Documentation — aturan plugin.path dan contoh connector FileStream bawaan distribusi.
• Debezium Documentation — Debezium Community — connector change data capture yang membaca log transaksi basis data.