Standalone vs Distributed Mode #

Apache Kafka Connect dirancang untuk melayani berbagai jenis skala kebutuhan pemindahan data, mulai dari pengujian lokal yang sederhana hingga pemrosesan jutaan event per detik di lingkungan produksi berskala perusahaan. Untuk memenuhi kebutuhan tersebut, Kafka Connect menyediakan dua mode runtime operasional yang berbeda: Standalone Mode dan Distributed Mode. Pemilihan mode runtime ini akan menentukan bagaimana koordinasi antar proses dijalankan, bagaimana konfigurasi connector didefinisikan, serta bagaimana toleransi kesalahan (fault tolerance) dan skalabilitas horizontal dikelola. Artikel ini akan membedah secara mendalam perbedaan arsitektural antara Standalone dan Distributed Mode, mengupas peran penting tiga topik internal Kafka Connect dalam memelihara konsistensi kluster, serta mendiskusikan mekanisme rebalancing dinamis yang menjaga pipa data tetap berjalan tanpa hambatan.

Standalone Mode: Karakteristik dan Use Case #

Standalone Mode adalah model runtime paling sederhana dari Kafka Connect. Pada mode ini, seluruh eksekusi framework Connect—termasuk pendefinisian connector, pembagian task, koordinasi pengerjaan, dan serialisasi data—dijalankan di dalam satu proses JVM (Java Virtual Machine) tunggal pada satu mesin server fisik atau virtual.

Cara Kerja dan Konfigurasi #

Pada Standalone Mode, kita tidak berinteraksi dengan Kafka Connect menggunakan REST API untuk membuat atau memodifikasi connector. Sebagai gantinya, semua konfigurasi didefinisikan menggunakan berkas properti teks lokal (.properties) yang dibaca saat proses worker pertama kali dinyalakan.

Perintah terminal untuk menjalankan worker standalone biasanya berbentuk seperti ini:

# Menjalankan worker standalone dengan satu file konfigurasi worker dan satu atau lebih file konfigurasi connector
connect-standalone.sh connect-standalone.properties mysql-source.properties elasticsearch-sink.properties

Di dalam berkas connect-standalone.properties, kita mendefinisikan konfigurasi worker dasar seperti detail koneksi broker Kafka dan Converter yang digunakan. Sementara berkas properti lainnya memuat parameter spesifik connector.

Keuntungan Standalone Mode #

• Sederhana dan Cepat: Sangat mudah untuk disiapkan karena tidak membutuhkan pembuatan topik internal khusus di Kafka.
• Ringan: Menggunakan memori heap JVM yang relatif kecil karena tidak ada overhead koordinasi antar mesin.
• Sempurna untuk Pengujian lokal: Sangat ideal bagi tim pengembang untuk melakukan debugging custom connector secara lokal di komputer mereka.

Kelemahan dan Batasan #

• Tidak Ada Toleransi Kesalahan (No High Availability): Jika proses JVM mengalami crash (misalnya karena OutOfMemory atau mesin mati), seluruh pipa data akan langsung berhenti. Failover harus dilakukan secara manual dengan menyalakan ulang service.
• Skalabilitas Terbatas: Paralelisme hanya dibatasi oleh kekuatan komputasi (CPU/Memory) dari satu mesin tunggal. Kita tidak bisa menambah mesin baru untuk membagi beban kerja secara otomatis.
• Manajemen Konfigurasi Statis: Setiap kali kita ingin menambah, menghapus, atau mengubah konfigurasi connector, kita harus mematikan worker, mengedit berkas properti, dan menyalakannya kembali (downtime).

Skenario Penggunaan yang Tepat #

Meskipun tidak cocok untuk lingkungan produksi utama, Standalone Mode sangat andal digunakan untuk:

• Mengalirkan file log lokal dari tepi jaringan (edge node) ke Kafka, mirip dengan agen log pengumpul (seperti Filebeat atau Fluentbit).
• Proses migrasi data satu kali (one-time data migration) di mana ketersediaan tinggi (high availability) bukan merupakan syarat kritis.

Distributed Mode: Desain Arsitektur Multi-Worker #

Untuk memenuhi kebutuhan sistem produksi modern yang menuntut zero-downtime, skalabilitas elastis, dan toleransi kesalahan otomatis, Kafka Connect menyediakan Distributed Mode. Pada mode ini, beberapa worker Connect (proses JVM indeks independen) dijalankan di beberapa server berbeda untuk membentuk satu kluster logis yang terpadu.

flowchart TD
    subgraph ClusterGroup ["Kafka Connect Distributed Cluster (group.id = connect-prod)"]
        WorkerA["Worker Node A (REST API: 8083)"]
        WorkerB["Worker Node B (REST API: 8083)"]
        WorkerC["Worker Node C (REST API: 8083)"]
    end
    
    subgraph InternalTopics ["Kafka Broker (Shared State Storage)"]
        ConfigTopic[("connect-configs (1 Partisi, Compacted)")]
        OffsetTopic[("connect-offsets (Compacted)")]
        StatusTopic[("connect-status (Compacted)")]
    end
    
    Dev["Developer / Admin API Client"] -->|REST Request: Create Connector| WorkerA
    WorkerA -->|Tulis Konfigurasi Baru| ConfigTopic
    ConfigTopic -.->|Deteksi Perubahan via Consumer| WorkerB & WorkerC
    
    WorkerB -->|Sinkronisasi Offset Tulis/Baca| OffsetTopic
    WorkerC -->|Publikasi Status Kesehatan| StatusTopic

1. Koordinasi Tanpa Master (Masterless Architecture) #

Kluster Distributed Mode tidak mengandalkan arsitektur Master-Worker tradisional yang memiliki satu titik kegagalan tunggal (single point of failure). Sebagai gantinya, koordinasi antar worker didasarkan pada Group Coordinator internal Apache Kafka, memanfaatkan protokol keanggotaan group (consumer group protocol) yang sama seperti yang digunakan oleh konsumen Kafka biasa.

Setiap worker yang menyala dengan konfigurasi group.id yang sama akan otomatis bergabung ke dalam kluster Connect yang sama. Salah satu worker akan ditunjuk secara dinamis oleh Kafka sebagai Leader kelompok, yang bertugas membagi tugas (task) ke seluruh worker yang aktif secara merata. Jika worker leader tersebut mati, Kafka secara instan akan menunjuk worker lain untuk mengambil alih peran kepemimpinan.

2. Manajemen Dinamis via REST API #

Di Distributed Mode, kita tidak lagi menyertakan berkas konfigurasi connector lokal saat menyalakan worker. Kita cukup menyalakan worker kosong:

# Menyalakan worker distributed hanya dengan berkas properti worker
connect-distributed.sh connect-distributed.properties

Setelah kluster worker menyala, semua operasi penambahan, modifikasi, pemantauan, dan penghapusan connector dilakukan secara dinamis melalui protokol HTTP REST API yang diekspos oleh setiap worker (secara default pada port 8083).

Sebagai contoh, untuk membuat MySQL Source Connector baru, kita cukup mengirimkan request HTTP POST dengan payload JSON:

curl -X POST -H "Content-Type: application/json" \
     --data '{
       "name": "mysql-source-dynamic",
       "config": {
         "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
         "tasks.max": "3",
         "connection.url": "jdbc:mysql://mysql-db:3306/shop",
         "connection.user": "read_user",
         "connection.password": "pass",
         "mode": "incrementing",
         "incrementing.column.name": "id",
         "topic.prefix": "shop-orders-"
       }
     }' http://localhost:8083/connectors

REST API ini mempermudah tim DevOps untuk mengintegrasikan deployment connector ke dalam pipa CI/CD otomatis tanpa perlu melakukan intervensi manual pada sistem operasi worker.

REST API Endpoint Reference untuk Operasi Kluster #

Distributed Mode mengekspos endpoint REST API yang lengkap pada setiap worker. Klien API (seperti curl, Postman, atau script automasi) dapat menghubungi IP worker mana saja di dalam kluster karena konfigurasi disinkronkan ke seluruh node secara real-time.

Berikut adalah daftar endpoint REST API yang wajib kita ketahui untuk mengoperasikan kluster Connect:

• GET /connectors

  • Kegunaan: Menampilkan daftar nama semua connector yang aktif di kluster.
  • Contoh Respons: ["mysql-source-dynamic", "elasticsearch-sink-logs"]

• POST /connectors

  • Kegunaan: Membuat connector baru. Payload berupa objek JSON yang berisi nama connector dan properti konfigurasinya.

• GET /connectors/{name}

  • Kegunaan: Mengambil detail konfigurasi aktif dari connector tertentu.

• GET /connectors/{name}/status

  • Kegunaan: Memeriksa kesehatan connector dan status task-task-nya secara real-time. Ini adalah endpoint paling penting untuk monitoring sistem.
  • Contoh Respons:

    {
      "name": "mysql-source-dynamic",
      "connector": {
        "state": "RUNNING",
        "worker_id": "192.168.1.50:8083"
      },
      "tasks": [
        {
          "id": 0,
          "state": "RUNNING",
          "worker_id": "192.168.1.50:8083"
        },
        {
          "id": 1,
          "state": "FAILED",
          "trace": "java.sql.SQLException: Connection pool exhausted...",
          "worker_id": "192.168.1.51:8083"
        }
      ],
      "type": "source"
    }
    

• PUT /connectors/{name}/config

  • Kegunaan: Mengubah konfigurasi connector yang sudah ada secara dinamis. Perubahan konfigurasi ini akan otomatis memicu rebalance task untuk menerapkan setelan baru.

• POST /connectors/{name}/restart

  • Kegunaan: Menyalakan ulang Connector instance (misalnya setelah konfigurasi diubah secara manual di sisi eksternal).

• POST /connectors/{name}/tasks/{task_id}/restart

  • Kegunaan: Menyalakan ulang task tertentu yang berstatus FAILED akibat gangguan transien tanpa menghentikan task lain yang berjalan normal.

• DELETE /connectors/{name}

  • Kegunaan: Menghapus connector secara permanen dan membebaskan seluruh task yang berjalan di bawahnya.

Pembedahan Parameter Konfigurasi Distributed Worker #

Untuk mengonfigurasi Distributed Worker secara optimal di lingkungan produksi, kita perlu mendalami berkas properti worker (connect-distributed.properties). Di bawah ini adalah konfigurasi parameter esensial beserta penjelasan fungsionalnya:

# ✓ BENAR: Mengonfigurasi bootstrap servers ke cluster broker Kafka produksi kita
bootstrap.servers=kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092

# Identitas kelompok kluster Connect. Semua worker dengan group.id yang sama membentuk 1 cluster
group.id=connect-cluster-production

# Port HTTP REST API yang dibuka untuk kueri administrasi
listeners=HTTP://0.0.0.0:8083
# Alamat host yang dipromosikan ke worker lain agar koordinasi internal REST berjalan normal
rest.advertised.host.name=connect-worker-1
rest.advertised.port=8083

# --- PENGATURAN TOPIK INTERNAL (Wajib Replikasi Tinggi) ---

# Topik untuk konfigurasi connector. Wajib disetel 1 partisi!
config.storage.topic=connect-configs-topic
config.storage.replication.factor=3

# Topik untuk offset Source Connector. Setel partisi agak banyak (misal 25)
offset.storage.topic=connect-offsets-topic
offset.storage.replication.factor=3
offset.storage.partitions=25

# Topik untuk status kesehatan task.
status.storage.topic=connect-status-topic
status.storage.replication.factor=3
status.storage.partitions=5

Peran Tiga Topik Internal Kafka Connect #

Salah satu pertanyaan arsitektural yang paling sering muncul adalah: Bagaimana kluster Distributed Mode menyimpan status dan konfigurasinya agar tetap konsisten di antara semua worker tanpa menggunakan database eksternal atau Zookeeper?

Jawabannya adalah: Kafka Connect menyimpan seluruh state kluster di dalam topik internal Apache Kafka. Ketika kluster worker pertama kali dijalankan, worker-worker tersebut akan otomatis membuat tiga topik internal berikut (jika belum ada) dengan konfigurasi retensi khusus:

1. Topik connect-configs #

Topik ini menyimpan konfigurasi lengkap dari semua connector yang telah didaftarkan ke kluster Connect.

• Konfigurasi Wajib: Harus disetel dengan 1 partisi saja (num.partitions=1) dan kebijakan cleanup compact (cleanup.policy=compact).
• Mengapa harus 1 partisi?: Hal ini mutlak diperlukan untuk menjamin urutan linearitas konfigurasi (total ordering of configuration changes). Semua worker Connect membaca topik ini dari awal hingga akhir untuk merekonstruksi daftar connector yang sama persis di memori lokal mereka.

2. Topik connect-offsets #

Topik ini digunakan oleh seluruh Source Connector yang berjalan di dalam kluster untuk menyimpan koordinat offset terakhir dari sistem sumber eksternal.

• Konfigurasi Wajib: Disetel dengan jumlah partisi yang lebih banyak (misalnya default 25 atau 50 partisi) untuk skalabilitas penulisan, dan menggunakan kebijakan cleanup.policy=compact.
• Cara Kerja: Setiap kali Source Task sukses mengirimkan batch data, ia akan menulis koordinat offset sumber ke topik ini. Jika task dipindahkan ke worker lain akibat rebalancing, task tersebut dapat melanjutkan pekerjaannya dari offset terakhir yang tercatat di topik ini.

3. Topik connect-status #

Topik ini merekam status kesehatan terbaru dari semua connector dan task yang sedang berjalan (apakah berstatus RUNNING, FAILED, PAUSED, dll.).

• Konfigurasi Wajib: Jumlah partisi banyak (misalnya default 5 atau 10 partisi) dengan kebijakan cleanup.policy=compact.
• Cara Kerja: Setiap kali status task berubah (misalnya crash akibat error), status terbaru dikirim ke topik ini. Hasil dari REST API kueri /connectors/{name}/status diambil langsung dari data terkompresi di topik ini.

[!WARNING] Di lingkungan produksi, ketiga topik internal ini wajib dibuat dengan faktor replikasi minimal 3 (replication.factor=3) dan jumlah minimum replika dalam sinkronisasi minimal 2 (min.insync.replicas=2). Kehilangan data pada topik connect-configs atau connect-offsets akan merusak seluruh status pipeline integrasi dan berpotensi menyebabkan kebocoran data duplikat yang masif.

Mekanisme Rebalancing dan Fault Tolerance #

Salah satu fitur paling unggul dari Distributed Mode adalah kemampuan toleransi kesalahan (fault tolerance) secara dinamis melalui mekanisme yang disebut Rebalancing.

Bagaimana Rebalance Terjadi? #

Ketika terjadi perubahan keanggotaan kluster Connect—misalnya, ada worker baru yang bergabung, worker lama mengalami crash (ditandai dengan hilangnya detak jantung / heartbeat timeout), atau ada konfigurasi connector baru yang didaftarkan—kluster akan memicu proses rebalancing.

Pada proses rebalancing:

1. Leader worker akan menghentikan distribusi tugas yang lama secara aman.
2. Leader menghitung ulang distribusi optimal berdasarkan jumlah worker yang aktif dan batas paralelisme tasks.max yang disetel pada konfigurasi connector.
3. Tugas-tugas (tasks) didistribusikan kembali secara merata ke seluruh worker yang tersedia.

Incremental Cooperative Rebalancing #

Pada versi Kafka Connect awal (sebelum versi 2.3), rebalancing menggunakan protokol Eager Rebalance yang bersifat memblokir (stop-the-world rebalance). Selama proses rebalancing berlangsung, seluruh task di kluster Connect akan dimatikan terlebih dahulu sebelum ditugaskan kembali. Hal ini menyebabkan lonjakan latensi yang mengganggu pada pipa data yang sensitif terhadap waktu.

Mulai versi 2.3 ke atas, Kafka Connect beralih menggunakan protokol Incremental Cooperative Rebalancing. Protokol baru ini sangat cerdas:

• Hanya task yang perlu dipindahkan (karena worker tempat ia berjalan mati) yang akan dihentikan.
• Task-task lain yang berjalan di worker yang sehat tetap dibiarkan berjalan normal tanpa interupsi.
• Hal ini secara dramatis mengurangi waktu downtime kluster dan meminimalkan duplikasi pesan akibat pemrosesan ulang dari offset lama.

Tabel Perbandingan Standalone vs Distributed #

Ringkasan #

• Standalone Mode — Pilihan terbaik untuk skenario local development, debugging, dan integrasi sederhana di satu mesin, menggunakan konfigurasi berbasis berkas lokal statis.
• Distributed Mode — Solusi produksi wajib yang menawarkan toleransi kesalahan otomatis (failover) dan skalabilitas horizontal elastis, dikelola dinamis menggunakan REST API.
• State via Kafka Topics — Konsistensi Distributed Mode dijaga sepenuhnya melalui tiga topik internal: connect-configs (1 partisi), connect-offsets, dan connect-status.
• Cooperative Rebalancing — Protokol modern Connect meminimalkan gangguan latensi dengan hanya memindahkan task yang terdampak kegagalan tanpa menghentikan pemrosesan task lain di kluster.
• Faktor Replikasi — Topik-topik internal Connect wajib disetel dengan faktor replikasi minimal 3 di lingkungan produksi guna melindungi integritas konfigurasi pipa data dari kegagalan broker.

  ← Sebelumnya: Source vs Sink Connector   Berikutnya: Integrasi Database & CDC →