Source vs Sink Connector #

Dalam ekosistem Apache Kafka, pemindahan data secara efisien antar platform membutuhkan pemahaman yang mendalam tentang arah aliran data. Kafka Connect menyederhanakan proses ini dengan menyediakan dua kategori utama plugin integrasi: Source Connector dan Sink Connector. Meskipun keduanya berjalan di bawah runtime worker yang sama dan menggunakan objek internal yang serupa, prinsip operasional, cara pengelolaan offset, toleransi kesalahan, serta penanganan laju data (backpressure) di antara keduanya sangat berbeda. Artikel ini akan membedah secara rinci perbandingan antara Source dan Sink Connector, menguraikan mekanisme internal masing-masing, membandingkan cara penyimpanan koordinat offset, serta mengupas bagaimana penanganan lonjakan data dilakukan secara asinkron di lingkungan produksi.

Konsep Dasar Source Connector #

Source Connector bertindak sebagai agen pengumpul data (ingestion agent) yang menarik informasi dari sistem eksternal dan mempublikasikannya ke dalam topik Apache Kafka. Sistem eksternal ini bisa berupa database transaksional (seperti PostgreSQL, Oracle, atau SQL Server), sistem antrean pesan legacy (seperti IBM MQ atau ActiveMQ), sistem file lokal, hingga aliran data real-time dari sensor IoT atau API pihak ketiga.

Alur Kerja Internal Source Connector #

Secara arsitektural, alur kerja penulisan data dari sistem eksternal ke Kafka melalui Source Connector dapat digambarkan sebagai rantai pemrosesan asinkron:

1. Polling Data: Task Source Connect berjalan dalam loop tak terbatas (infinite loop) yang memanggil metode poll() secara berkala untuk mengambil data dari sistem sumber.
2. Pembuatan SourceRecord: Setiap baris database, pesan antrean, atau baris log yang didapatkan kemudian diubah oleh kelas connector menjadi objek internal Java bernama SourceRecord.
3. Penerapan SMT (Single Message Transforms): Jika dikonfigurasi, objek SourceRecord dikirim ke rantai transformasi memori untuk dimodifikasi secara dinamis (seperti penggantian nama kolom atau penyaringan data).
4. Serialisasi via Converter: SourceRecord yang sudah bersih diterjemahkan oleh Converter (misalnya AvroConverter atau JsonConverter) menjadi representasi biner (byte array).
5. Publish ke Broker: Worker Connect bertindak sebagai Kafka Producer yang mengirimkan payload biner tersebut ke broker Kafka menggunakan Producer API standar.

Tujuan utama dari Source Connector adalah memastikan bahwa setiap data yang baru muncul di sistem sumber terkirim ke Kafka dengan latensi sekecil mungkin tanpa membebani sistem sumber itu sendiri.

Konsep Dasar Sink Connector #

Sebaliknya, Sink Connector bertindak sebagai agen penyalur data (egress agent) yang membaca data dari satu atau beberapa topik Apache Kafka dan menulisnya ke dalam sistem penyimpanan eksternal. Sistem target ini biasanya berupa database pencarian (Elasticsearch/OpenSearch), sistem analisis data besar (ClickHouse, Snowflake, Google BigQuery), penyimpanan objek cloud (AWS S3, Azure Blob Storage), atau database operasional relasional/NoSQL.

Alur Kerja Internal Sink Connector #

Sink Connector beroperasi dengan arah yang berlawanan dari Source Connector, memanfaatkan fungsionalitas konsumsi standar Kafka:

1. Consumer Fetch: Worker Connect bertindak sebagai Kafka Consumer yang berlangganan (subscribe) pada topik-topik tertentu. Worker memanggil metode poll() Kafka Consumer API untuk menarik byte biner dari broker.
2. Deserialisasi via Converter: Byte biner yang ditarik dari broker Kafka diterjemahkan oleh Converter kembali menjadi objek internal Java bernama SinkRecord yang berisi skema dan nilai data asli.
3. Penerapan SMT: SMT memproses objek SinkRecord dalam memori untuk melakukan manipulasi atau penambahan data sebelum diserahkan ke connector.
4. Buffer & Batching: Task Sink Connector mengumpulkan objek-objek SinkRecord ke dalam memori buffer lokal guna melakukan pengelompokkan penulisan (batching) agar tidak membebani sistem target dengan transaksi kecil yang terlalu sering.
5. Write ke Target: Task memanggil metode put() untuk menulis kumpulan record data tersebut ke sistem penyimpanan eksternal (misalnya melakukan query SQL INSERT massal atau mengirimkan dokumen via Elasticsearch Bulk HTTP API).

Dalam mode Sink, fokus utama arsitektur adalah menjaga kontinuitas pengiriman data dan menangani kegagalan sistem target tanpa merusak urutan pembacaan offset Kafka.

Perbandingan Pengelolaan Offset (Offset Management) #

Salah satu perbedaan paling mendasar antara Source dan Sink Connector terletak pada bagaimana mereka mengelola status kemajuan kerja mereka (offset management). Offset adalah koordinat penunjuk posisi data terakhir yang telah sukses diproses.

flowchart TD
    subgraph SourceOffsetTracking ["Source Offset Tracking"]
        direction TB
        SourceDB[("Sistem Sumber")] -->|Baca data di pos X| SrcTask["Source Task"]
        SrcTask -->|Kirim record + Metadata Pos X| WorkerProducer["Worker Producer"]
        WorkerProducer -->|Tulis data| DataTopic(("Kafka Data Topic"))
        WorkerProducer -->|Catat Pos X secara asinkron| OffsetTopic(("connect-offsets Topic"))
    end
    
    subgraph SinkOffsetTracking ["Sink Offset Tracking"]
        direction TB
        ConsumerPoll["Worker Consumer"] -->|Tarik data dari Offset Y| DataTopic
        ConsumerPoll -->|Kirim SinkRecord| SnkTask["Sink Task"]
        SnkTask -->|Tulis sukses ke target| TargetStorage["Penyimpanan Target"]
        TargetStorage -.->|Verifikasi sukses| SnkTask
        SnkTask -->|Komit Offset Y ke Kafka| CommitTopic(("__consumer_offsets Topic"))
    end

1. Pengelolaan Offset pada Source Connector #

Pada Source Connector, Kafka tidak mengetahui struktur data internal sistem sumber. Sebagai contoh, jika kita membaca baris tabel database, koordinat offsetnya mungkin berupa nilai kolom incrementing_id atau timestamp last_modified. Jika kita membaca file teks, offsetnya adalah indeks byte posisi kursor file.

• Format Offset Kustom: Kafka Connect memungkinkan Source Connector mendefinisikan skema offset mereka sendiri dalam format key-value. Key mendefinisikan partisi sumber (misalnya {"table": "orders"} atau {"file": "logs.txt"}), dan Value mendefinisikan posisi offset sumber (misalnya {"id": 1024} atau {"byte_offset": 4560}).
• Tempat Penyimpanan: Koordinat offset kustom ini secara berkala ditulis oleh worker Connect secara asinkron ke dalam topik internal khusus di Kafka yang bernama connect-offsets. Kebijakan pembersihan log (cleanup policy) pada topik ini disetel ke compact agar posisi terakhir setiap partisi sumber selalu terjaga secara persisten.

2. Pengelolaan Offset pada Sink Connector #

Pada Sink Connector, koordinat offsetnya adalah offset partisi Kafka standar (angka integer linier 64-bit yang menunjukkan posisi pesan di dalam partisi topik Kafka).

• Format Offset Standar: Offset yang digunakan adalah pasangan Topic-Partition-Offset standar Kafka.
• Tempat Penyimpanan: Karena worker Connect bertindak sebagai konsumen Kafka biasa, offset konsumen dikelola langsung oleh broker Kafka melalui mekanisme Group Coordinator dan disimpan di dalam topik sistem internal __consumer_offsets.
• Sinkronisasi Commit: Komit offset ke broker Kafka hanya akan dilakukan setelah Sink Task mengonfirmasi bahwa seluruh record dalam batch pemanggilan put() terbaru telah sukses ditulis ke sistem target eksternal. Jika sistem target gagal menerima data (misalnya terjadi timeout jaringan), offset tidak akan dikomit, dan record yang sama akan dikirim ulang oleh Kafka untuk diproses kembali (At-Least-Once delivery semantics).

Detail Mekanisme Serialisasi dan Deserialisasi dalam Flow Source & Sink #

Untuk memahami bagaimana data dimanipulasi secara fisik, kita perlu melihat bagaimana objek-objek biner diproses pada level converter. Model decoupling di Kafka Connect menjamin bahwa penulisan biner dan logika pembacaan skema berjalan secara modular.

1. Perjalanan Data di Sisi Source #

Ketika Source Task menarik baris tabel database, task tersebut merakit struktur data menjadi objek SourceRecord.

• Tahap 1: Java Heap representation: Objek ini membawa Schema terdefinisi (seperti tipe data VARCHAR dipetakan ke string Java, INT dipetakan ke Integer) dan Value aktual (seperti "Rudi" dan 25).

• Tahap 2: Serialisasi oleh Converter: Jika kita menyetel value.converter=org.apache.kafka.connect.json.JsonConverter dengan schemas.enable=true, maka output payload biner yang dikirim ke Kafka akan memuat seluruh struktur metadata skema di setiap pesan:

  {
    "schema": {
      "type": "struct",
      "fields": [
        {"type": "string", "optional": false, "field": "name"},
        {"type": "int32", "optional": true, "field": "age"}
      ],
      "optional": false
    },
    "payload": {
      "name": "Rudi",
      "age": 25
    }
  }
  

  Ini adalah anti-pattern untuk arsitektur skala besar karena data skema diulang terus-menerus dan memboroskan ruang disk hingga 90%.

  Sebaliknya, jika kita menggunakan AvroConverter, skema didaftarkan ke Confluent Schema Registry. Skema disimpan sekali di registry, dan biner yang terkirim ke Kafka hanya berukuran: [1-byte Magic Byte] + [4-byte Schema ID] + [Biner Payload Avro (tanpa nama kolom)] Hal ini menghemat bandwidth jaringan dan memori heap secara dramatis.

2. Perjalanan Data di Sisi Sink #

Ketika Sink Task menerima pesan dari broker Kafka:

• Tahap 1: Pembacaan Biner: Worker Connect mengambil byte mentah dari topik Kafka.
• Tahap 2: Deserialisasi oleh Converter: Converter mendeteksi format data. Jika menggunakan AvroConverter, ia membaca Schema ID dari 5 byte pertama pesan, mengunduh skema tersebut dari Schema Registry (jika belum ada di memori cache lokal worker), lalu memecah biner menjadi objek Java SinkRecord.
• Tahap 3: Penyerahan ke Connector: Objek SinkRecord yang telah direkonstruksi diserahkan ke Sink Connector untuk ditulis ke database tujuan menggunakan sintaks yang kompatibel dengan database tersebut.

Penanganan Backpressure pada Sink Connector #

Backpressure adalah mekanisme pertahanan sistem terdistribusi untuk mengendalikan laju aliran data ketika sistem penerima (receiver) tidak mampu mengimbangi laju data yang dikirim oleh sistem pengirim (sender). Skenario ini sangat sering terjadi pada Sink Connector di mana kecepatan broker Kafka memancarkan pesan jauh melebihi kecepatan database target dalam menulis data ke disk.

Bagaimana Kafka Connect Menangani Backpressure? #

Secara default, consumer Kafka di dalam worker Connect menggunakan model tarik (pull model) yang secara inheren mendukung koordinasi backpressure yang sangat dinamis. Ketika database eksternal mengalami kelambatan (misalnya database PostgreSQL sedang menjalankan proses vacuuming intensif atau Elasticsearch mengalami pemakaian CPU 100%):

1. Penundaan Eksekusi put(): Task Sink Connector akan mendeteksi perlambatan tulis karena koneksi thread atau pool database tertahan menunggu respons. Akibatnya, pemanggilan fungsi put() di dalam task Connect akan memakan waktu lebih lama untuk diselesaikan.
2. Penahanan Polling Partisi: Worker Connect memantau kesehatan eksekusi task. Jika buffer antrean internal untuk task tersebut penuh karena eksekusi put() sebelumnya belum selesai, worker akan secara asinkron memanggil API suspensi konsumen Kafka, yaitu KafkaConsumer.pause(Collection<TopicPartition> partitions).
3. Penghentian Sementara Fetching: Broker Kafka akan berhenti mengirimkan data baru untuk partisi yang sedang di-pause tersebut. Tidak ada alokasi memori heap JVM tambahan yang terbuang untuk menyimpan pesan yang mengantre.
4. Resume Polling: Begitu database target kembali normal dan task berhasil menyelesaikan penulisan batch data yang tertunda, worker akan memanggil KafkaConsumer.resume(Collection<TopicPartition> partitions) untuk mulai menarik data dari Kafka kembali secara normal.

Parameter Penting untuk Tuning Backpressure #

Untuk mengoptimalkan ketahanan Sink Connector dari crash akibat kehabisan memori (OutOfMemoryError) saat menghadapi lonjakan data, kita harus mengonfigurasi parameter konsumsi pada berkas properti connector secara cermat:

# Jumlah maksimum record yang ditarik dalam sekali pemanggilan poll()
# Nilai default: 500. Perkecil jika payload per pesan sangat besar.
consumer.max.poll.records=200

# Batas waktu maksimal bagi task untuk memproses data sebelum dianggap mati
# Jika database target lambat menulis data dan melebihi batas ini, 
# broker akan menganggap worker Connect mati dan memicu proses rebalancing group.
consumer.max.poll.interval.ms=300000

# Waktu tunggu koneksi HTTP/Socket ke sistem target sebelum melempar Exception timeout
connection.timeout.ms=10000

Jika database target sering kali membutuhkan waktu penulisan batch melebihi 5 menit, kita harus menaikkan consumer.max.poll.interval.ms agar cluster Connect tidak mengalami rebalance storm yang sia-sia akibat task dianggap mati secara keliru.

Skenario Penanganan Kegagalan (Failure Handling Scenario) #

Sistem terdistribusi rentan terhadap kegagalan komponen. Pemisahan peran antara Source dan Sink Connector menentukan bagaimana penanganan kegagalan (fault tolerance) dilakukan.

1. Kegagalan pada Source Connector #

• Masalah: Koneksi database terputus, query polling timeout, atau kredensial database kedaluwarsa.
• Mekanisme Penanganan:
  • Task Source akan mencoba menghubungi database kembali jika dikonfigurasi dengan toleransi koneksi. Jika gagal total, task akan masuk ke status FAILED.
  • Karena data di database sumber bersifat persisten, tidak ada data yang hilang selama downtime task.
  • Setelah masalah koneksi diperbaiki, kita dapat mengirimkan perintah REST API POST /connectors/{name}/restart untuk menyalakan kembali task. Task akan membaca offset terakhir dari topik connect-offsets dan melanjutkan pemindahan data dengan aman.

2. Kegagalan pada Sink Connector #

• Masalah: Penyimpanan target Elasticsearch penuh (disk space penuh), otentikasi API key kedaluwarsa, atau format data melanggar aturan skema tabel database SQL (misalnya constraint violation).
• Mekanisme Penanganan:
  • Jika terjadi kesalahan biner/jaringan, task akan melempar exception dan berhenti. Offset Kafka tidak dikomit ke broker.
  • Jika masalah disebabkan oleh format pesan yang rusak (poison pill), menyetop seluruh pipeline akan merugikan data lain yang valid. Untuk itu, kita perlu mengonfigurasi toleransi kesalahan khusus di sisi Sink:

    # Meneruskan pemrosesan meskipun dijumpai record rusak
    errors.tolerance=all
    # Mengirim record rusak ke Dead Letter Queue (DLQ)
    errors.deadletterqueue.topic.name=orders-sink-dlq
    errors.deadletterqueue.context.headers.enable=true
    

  • Hal ini memastikan data yang valid tetap terkirim ke target, sementara data yang bermasalah dipisahkan ke topik DLQ untuk dianalisis oleh tim pengembang tanpa memacetkan proses bisnis.

Contoh Konfigurasi Riil dan Pembedahan Parameter #

Untuk memberikan pemahaman yang komprehensif, mari kita bedah contoh berkas konfigurasi untuk masing-masing tipe connector.

1. Contoh Konfigurasi Source Connector (MySQL JDBC) #

Di bawah ini adalah konfigurasi MySQL Source Connector untuk menarik data transaksi ke Kafka:

{
  "name": "mysql-sales-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:mysql://mysql-server:3306/sales_db",
    "connection.user": "connect_user",
    "connection.password": "Password123!",
    "table.whitelist": "orders,customers",
    
    "//": "Menggunakan kolom incrementing ID untuk mendeteksi baris baru",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    
    "//": "Setiap tabel akan ditulis ke topik terpisah dengan prefix mysql-db-",
    "topic.prefix": "mysql-db-",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

• Pembedahan Parameter:
  • tasks.max: Menentukan jumlah thread paralel (maksimum 2). Karena ada 2 tabel (orders dan customers), setiap task akan bertugas membaca 1 tabel secara paralel.
  • mode: Penyetelan incrementing memberi tahu connector untuk melacak baris baru berdasarkan kenaikan nilai kunci utama (id). Koordinat ID terbesar terakhir akan terus dicatat di topik connect-offsets.

2. Contoh Konfigurasi Sink Connector (Elasticsearch Sink) #

Di bawah ini adalah konfigurasi Elasticsearch Sink Connector untuk menulis data dari Kafka ke Elasticsearch index:

{
  "name": "elasticsearch-sales-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "3",
    "topics": "mysql-db-orders",
    "connection.url": "http://elasticsearch-node:9200",
    
    "//": "Menggunakan offset Kafka untuk menjaga keunikan dokumen di ES",
    "key.ignore": "false",
    "schema.ignore": "true",
    
    "//": "Menyetel batching untuk mengoptimalkan performa I/O Elasticsearch",
    "batch.size": "1000",
    "flush.timeout.ms": "5000"
  }
}

• Pembedahan Parameter:
  • tasks.max: Dapat disetel hingga 3 jika topik mysql-db-orders memiliki 3 partisi atau lebih. Jika partisi topik hanya 1, maka task ke-2 dan ke-3 akan menganggur (idle).
  • key.ignore: Penyetelan false berarti kunci pesan Kafka akan digunakan sebagai ID dokumen Elasticsearch. Ini penting untuk mencegah duplikasi data jika terjadi pengiriman ulang pesan akibat crash jaringan (idempotent updates).

Tabel Perbandingan Komprehensif Source vs Sink #

Ringkasan #

• Arah Aliran — Source Connector bertindak sebagai produser data yang menarik informasi dari luar ke Kafka, sedangkan Sink Connector bertindak sebagai konsumen data yang mendorong data dari Kafka ke luar.
• Offset Kustom vs Standar — Source Connector menggunakan offset kustom key-value yang disimpan di topik connect-offsets, sementara Sink Connector menggunakan offset partisi standar di topik __consumer_offsets.
• Mekanisme Commit — Sink Connector hanya melakukan commit offset setelah sistem target mengonfirmasi kesuksesan penulisan data guna menjaga jaminan At-Least-Once.
• Otomatisasi Backpressure — Sink Connector menggunakan model tarik Kafka Consumer dan API pause/resume untuk menghentikan aliran data sementara ketika sistem target mengalami overload.
• Tuning Parameter — Parameter consumer.max.poll.interval.ms dan consumer.max.poll.records harus disesuaikan dengan karakteristik throughput sistem target untuk mencegah restart task yang tidak perlu.

  ← Sebelumnya: Apa Itu Kafka Connect?   Berikutnya: Standalone vs Distributed Mode →