Error Handling & DLQ #

Dalam sistem terdistribusi berskala besar yang mengalirkan jutaan event setiap detiknya, kesalahan (error) bukanlah pertanyaan “apakah akan terjadi” melainkan “kapan terjadi”. Di dalam pipa data Kafka Connect, kesalahan dapat muncul dari berbagai sumber, mulai dari pemadaman jaringan sementara (network outage), kegagalan otentikasi, hingga masuknya pesan rusak yang formatnya tidak sesuai dengan skema (poison pills). Jika kita tidak merancang strategi penanganan kesalahan dengan matang, satu saja pesan rusak yang dikirim oleh produser dapat menyebabkan seluruh cluster worker Connect crash dan menghentikan aliran bisnis penting kita. Untuk mengantisipasi hal tersebut, Kafka Connect menyediakan fitur Error Handling bawaan dan Dead Letter Queue (DLQ). Artikel ini akan membedah klasifikasi jenis kegagalan di Kafka Connect, menjelaskan penyetelan toleransi kesalahan, menguraikan cara konfigurasi DLQ, serta membahas taktik penggulangan otomatis (retry) yang andal di produksi.

Jenis-Jenis Kegagalan di Kafka Connect (Classification of Failures) #

Untuk menangani kesalahan secara tepat, kita harus terlebih dahulu mengklasifikasikan jenis kegagalan yang terjadi di dalam pipeline Kafka Connect menjadi tiga kategori utama:

1. Kegagalan Jaringan Sementara (Transient Network Failures) #

Kegagalan ini bersifat temporer dan biasanya disebabkan oleh kendala infrastruktur jaringan atau sistem eksternal yang sedang sibuk.

Contoh: Timeout koneksi database, sistem target sedang restart, atau partisi jaringan sementara (network partition) antara worker Connect dengan broker Kafka.
Solusi: Masalah ini dapat diselesaikan secara otomatis dengan melakukan pengulangan pengiriman (retry) setelah jeda waktu tertentu.

2. Kesalahan Format Data (Poison Pills / Serialization Errors) #

Kesalahan ini bersifat permanen pada record data tertentu. Record tersebut tidak akan pernah bisa didecode karena rusak secara struktural.

Contoh: Konsumen mengharapkan format Avro biner, namun produser mengirimkan string teks acak, atau JSON yang dikirim kehilangan field wajib (non-nullable constraint).
Solusi: Melakukan retry pada data yang sama tidak akan menyelesaikan masalah. Pesan rusak ini harus diabaikan atau dipisahkan ke wadah karantina (DLQ) agar pipeline tetap dapat memproses data valid berikutnya.

3. Kesalahan Logika Sistem Target (Target System Logic Errors) #

Kesalahan yang dilemparkan oleh sistem penyimpanan eksternal saat proses penulisan.

Contoh: Melakukan operasi SQL INSERT yang melanggar aturan kunci unik (unique constraint violation) pada tabel database target, atau mengirim dokumen dengan struktur field yang bertentangan dengan indeks Elasticsearch.
Solusi: Memerlukan penanganan khusus di sisi connector (seperti menyetel operasi ke mode upsert untuk mengabaikan duplikasi key).

Strategi Toleransi Kesalahan (errors.tolerance) #

Kafka Connect menyediakan parameter konfigurasi penting bernama errors.tolerance untuk menentukan sikap worker ketika mendeteksi adanya record data yang rusak atau gagal diproses pada tahap Converter, SMT, atau penulisan Connector.

Ada dua opsi nilai yang dapat kita setel pada properti ini:

flowchart TD
    R1["Record Data Rusak"] -->|errors.tolerance = none| T1["Task CRASH / STOP (Fail-Fast)"]
    R2["Record Data Rusak"] -->|errors.tolerance = all| T2["Log Error / Kirim ke DLQ (Tolerant)"]

1. `errors.tolerance = none` (Default / Fail-Fast) #

Ini adalah setelan bawaan dari Kafka Connect. Jika terjadi kesalahan sekecil apa pun pada satu record data, task Connect akan langsung melempar Exception, menghentikan eksekusi thread-nya, dan mengubah statusnya menjadi FAILED.

Kapan Digunakan: Sangat disarankan untuk pipa data keuangan (financial transaction), billing, atau sistem kritis lain yang menuntut konsistensi data 100%. Lebih baik sistem berhenti total daripada menulis data yang salah atau mengalami kehilangan data (zero data loss guarantee).

2. `errors.tolerance = all` (Tolerant Mode) #

Dalam mode ini, jika terjadi kesalahan pemrosesan pada record data, worker Connect akan mencatat log kesalahan tersebut, mengabaikan record rusak tersebut, dan langsung melanjutkan pemrosesan ke record berikutnya tanpa menghentikan task.

Kapan Digunakan: Cocok untuk sistem analisis log telemetri, metrik IoT, atau pelacakan aktivitas web (clickstream) di mana kehilangan beberapa baris data tidak memengaruhi hasil analitik secara signifikan.
Risiko: Mengaktifkan properti ini tanpa DLQ akan menyebabkan data rusak hilang selamanya tanpa disadari oleh tim pengembang.

Bagaimana dengan Source Connector? (Handling Source Errors) #

Seperti yang telah dibahas, Dead Letter Queue bawaan Kafka Connect hanya bekerja untuk Sink Connector. Lantas, bagaimana cara kita mengelola record data yang rusak di sisi Source Connector sebelum sempat ditulis ke Kafka?

Jika Source Connector menjumpai baris data database yang korup (misalnya data teks biner yang tidak bisa di-parse oleh converter):

Pola Custom SMT Routing: Kita dapat menulis plugin Single Message Transform (SMT) kustom. SMT ini dibungkus dengan blok try-catch. Jika parsing payload gagal, SMT akan mengubah nama topik tujuan record tersebut ke topik khusus (misal corrupted-database-records) alih-alih melempar Exception yang menghentikan task.
Filter Drop: Jika data rusak di sumber boleh diabaikan, SMT kustom dapat menyaring data tersebut dengan mengembalikan nilai null, sehingga worker Connect tidak akan mengirimkan data tersebut ke broker Kafka.

Konfigurasi Dead Letter Queue (DLQ) untuk Record Rusak #

Untuk menyeimbangkan antara kelangsungan pipa data (agar task tidak crash) dan keamanan data (agar data rusak tidak hilang), kita harus memadukan "errors.tolerance": "all" dengan Dead Letter Queue (DLQ).

Dead Letter Queue adalah topik Kafka khusus yang digunakan untuk menampung setiap record data yang gagal diproses oleh Sink Connector.

flowchart TD
    subgraph ConnectProcess ["Connect Task Processing Loop"]
        Record["Record dari Kafka Topic"] -->|1. Baca| Converter{"Converter / SMT"}
        Converter -->|Sukses| Process["Kirim ke Target System"]
        Converter -->|Gagal / Rusak| ErrorCheck{"errors.tolerance?"}
        
        ErrorCheck -->|none| Crash["Task Crash & Stop"]
        ErrorCheck -->|all| DLQCheck{"DLQ Dikonfigurasi?"}
        
        DLQCheck -->|Ya| WriteDLQ["Tulis ke DLQ Topic + Headers"]
        DLQCheck -->|Tidak| Skip["Abaikan & Buang Record"]
        
        WriteDLQ & Skip & Process -->|Lanjut Loop| Record
    end

Menyimpan Metadata Error ke Message Headers #

Saat Kafka Connect mengalihkan pesan rusak ke topik DLQ, pesan tersebut tidak hanya dikirim apa adanya. Kita dapat mengonfigurasi worker agar menyisipkan metadata kesalahan yang sangat berharga ke dalam Headers pesan Kafka Connect tersebut.

Metadata yang disisipkan meliputi:

deadletterqueue.topic: Nama topik asli asal pesan rusak.
deadletterqueue.partition: ID partisi asli.
deadletterqueue.offset: Posisi offset asli pesan yang bermasalah.
deadletterqueue.reason: Pesan exception log mengapa record ini gagal diproses.
deadletterqueue.error.class: Nama kelas Exception Java yang terlempar.

Dengan adanya header metadata ini, tim pengembang dapat dengan mudah membuat aplikasi pemonitor yang membaca topik DLQ, menganalisis penyebab kerusakan, memperbaiki data, dan mengirimkannya kembali (re-drive) ke pipeline utama secara otomatis.

Mekanisme Log Kesalahan Kustom (Custom Error Logging & Alerting) #

Selain menyaring data rusak ke DLQ, sistem produksi yang andal harus memiliki visibilitas penuh terhadap kapan dan mengapa kesalahan terjadi.

1. Konfigurasi Log4j pada Worker Connect #

Kita dapat mengontrol seberapa detail log kesalahan dicatat ke dalam berkas log sistem. Di dalam file konfigurasi log4j worker (/opt/kafka/config/connect-log4j.properties), kita harus mengonfigurasi package penanganan error khusus:

# ✓ BENAR: Mengaktifkan detail debug untuk melacak pelacakan stacktrace error Connect
log4j.logger.org.apache.kafka.connect.runtime.WorkerSinkTask=DEBUG
log4j.logger.org.apache.kafka.connect.runtime.errors=DEBUG

Log ini nantinya akan dikirim ke sistem log terpusat (seperti Elasticsearch, Splunk, atau Datadog) untuk memicu peringatan (alerting) real-time ke Slack atau PagerDuty ketika jumlah Exception melonjak di atas ambang batas normal.

2. Monitoring via Metrik JMX Error #

Kafka Connect mengekspos metrik penanganan error spesifik yang wajib kita integrasikan dengan dashboard Prometheus/Grafana:

total-record-errors: Jumlah record yang gagal diproses (mengalami error) sejak task dijalankan.
total-record-failures: Jumlah record yang memicu kegagalan task (menyebabkan status task FAILED).
total-records-skipped: Jumlah record rusak yang diabaikan/dilewati karena setelan errors.tolerance=all.
deadletterqueue-produce-failures: Jumlah kegagalan saat worker mencoba menulis pesan rusak ke topik DLQ (misalnya karena otorisasi ACL bermasalah).

Pola Re-drive Dead Letter Queue (DLQ Re-drive Patterns) #

Pesan rusak yang mendarat di DLQ tidak boleh dibiarkan menumpuk selamanya tanpa penanganan. Kita harus merancang pola Re-drive (proses pemrosesan kembali data DLQ):

flowchart TD
    DLQ["DLQ Kafka Topic"] --> App["Read via Re-drive App"]
    App --> B1["Jaringan / Downtime Target?"] -- "Fix Target" --> WD["Write Direct"]
    App --> B2["Format Data Rusak?"] -- "Fix Payload" --> RI["Re-ingest"]

1. Skenario Kegagalan Jaringan #

Jika data masuk ke DLQ karena database target downtime melebihi batas waktu retry:

Tim DevOps memperbaiki database target hingga kembali menyala normal.
Aplikasi re-drive sederhana membaca pesan dari DLQ dan menulisnya langsung ke database target menggunakan database client standar.

2. Skenario Format Payload Rusak #

Jika data masuk ke DLQ karena format payload rusak (misalnya kolom created_at bertipe data tidak valid):

Kita membuat microservice kecil (atau aplikasi Kafka Streams) yang berlangganan ke topik DLQ.
Aplikasi tersebut membaca pesan, memperbaiki format datanya (misalnya mem-parsing string tanggal yang salah ke format ISO standar), lalu mempublikasikan kembali pesan yang sudah bersih tersebut ke topik Kafka bisnis utama agar dikonsumsi ulang oleh Sink Connector.

Otomatisasi Restart Task yang Gagal (Auto-Recovery Script) #

Di lingkungan produksi yang dinamis, task-task Connect dapat gagal karena error jaringan transien yang melampaui batas waktu retry. Dibandingkan melakukan restart manual lewat perintah REST API setiap kali menerima alert, kita dapat mendeploy script otomasi auto-recovery sederhana yang berjalan sebagai cron job di cluster Connect:

#!/bin/bash
# Script otomatisasi untuk mendeteksi dan restart task Kafka Connect yang failed

CONNECT_URL="http://localhost:8083"

# Mengambil daftar semua connector aktif
CONNECTORS=$(curl -s "${CONNECT_URL}/connectors")

for connector in $(echo "${CONNECTORS}" | jq -r '.[]'); do
  # Mengecek status kesehatan connector dan task
  STATUS=$(curl -s "${CONNECT_URL}/connectors/${connector}/status")
  
  # Iterasi setiap task
  for task in $(echo "${STATUS}" | jq -r '.tasks[].id'); do
    STATE=$(echo "${STATUS}" | jq -r ".tasks[${task}].state")
    
    if [ "${STATE}" == "FAILED" ]; then
      echo "✗ Menemukan task ${task} pada connector ${connector} berstatus FAILED!"
      echo "Mencoba melakukan restart task secara otomatis..."
      
      # Mengirim perintah restart via REST API
      curl -s -X POST "${CONNECT_URL}/connectors/${connector}/tasks/${task}/restart"
      
      echo "✓ Perintah restart sukses dikirim untuk task ${task}!"
    fi
  done
done

Mekanisme Retry Otomatis (Automatic Retry Mechanisms) #

Untuk mengatasi kegagalan jaringan yang bersifat sementara (transien), kita tidak boleh langsung memindahkan pesan ke DLQ. Kita harus memberikan kesempatan bagi worker Connect untuk mencoba mengirim ulang data tersebut secara otomatis selama jangka waktu tertentu.

Kafka Connect Core menyediakan dua parameter utama untuk mengontrol retries:

errors.retry.timeout: Total durasi waktu maksimal (dalam milidetik) bagi task untuk terus mencoba mengirim ulang batch data yang gagal sebelum akhirnya menyerah dan melempar error. Default-nya adalah 0 (tanpa retry).
errors.retry.delay.max.ms: Waktu jeda maksimal (dalam milidetik) di antara pemanggilan retry. Connect menggunakan algoritma exponential backoff di mana jeda waktu akan terus meningkat di setiap kegagalan hingga mencapai batas maksimal yang disetel di properti ini (default: 60000 ms atau 1 menit).

Di bawah ini adalah contoh setelan toleransi retry yang andal untuk menghadapi downtime database target yang berlangsung singkat:

# Mencoba mengirim ulang data selama maksimal 5 menit (300.000 ms) sebelum task dinyatakan gagal
errors.retry.timeout=300000

# Jeda maksimal antar pengiriman ulang adalah 10 detik
errors.retry.delay.max.ms=10000

Contoh Konfigurasi Sink Connector Lengkap dengan Error Handling & DLQ #

Berikut adalah konfigurasi JSON lengkap untuk mendeploy PostgreSQL Sink Connector produksi yang dilengkapi penanganan error transien via retry, toleransi kesalahan format data, dan pengalihan poison pill ke topik DLQ lengkap dengan headers metadata:

{
  "name": "postgresql-secure-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "2",
    "topics": "customer-transactions",
    "connection.url": "jdbc:postgresql://postgres-db:5432/finance",
    "connection.user": "sink_writer",
    "connection.password": "${file:/etc/connect/secrets:pg_writer_password}",
    
    "//": "--- Konfigurasi Retries untuk Error Jaringan Transien ---",
    "//": "Terus mencoba mengirim ulang selama 3 menit",
    "errors.retry.timeout": "180000",
    "errors.retry.delay.max.ms": "5000",
    
    "//": "--- Konfigurasi Toleransi Kesalahan Format Data ---",
    "//": "Toleransi all agar task tidak crash saat menjumpai pesan JSON rusak",
    "errors.tolerance": "all",
    
    "//": "--- Konfigurasi Dead Letter Queue (DLQ) ---",
    "errors.deadletterqueue.topic.name": "dlq-customer-transactions",
    "//": "✓ BENAR: Mengaktifkan pencatatan headers metadata error di DLQ untuk debugging",
    "errors.deadletterqueue.context.headers.enable": "true",
    
    "//": "--- Pengaturan Tulis Database ---",
    "insert.mode": "insert",
    "auto.create": "false",
    
    "//": "--- Converters ---",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

Ringkasan #

Klasifikasi Kegagalan — Error di Kafka Connect terbagi menjadi kegagalan jaringan sementara (butuh retry), kesalahan format data / poison pills (butuh DLQ), dan kesalahan logika target.

Fail-Fast vs Tolerant — Gunakan errors.tolerance=none untuk sistem transaksi kritis finansial demi konsistensi mutlak, dan errors.tolerance=all untuk log/metrik non-kritis.

Dead Letter Queue — DLQ menyaring dan mengisolasi data rusak ke topik Kafka terpisah, menjaga kelangsungan operasional task distributed worker Connect.

Metadata via Headers — Aktifkan errors.deadletterqueue.context.headers.enable=true untuk menyisipkan kelas Exception dan lokasi offset asli ke header pesan DLQ guna mempermudah proses debugging.

Tuning Log4j & JMX — Pantau metrik JMX total-records-skipped dan sesuaikan package logging WorkerSinkTask ke level DEBUG untuk memperoleh visibilitas penuh saat error terjadi.

DLQ Re-drive — Bangun aplikasi pembaca DLQ berkala untuk memperbaiki kesalahan data (poison pills) dan mengirimkannya kembali (re-ingest) ke topik bisnis utama.

Otomatisasi Restart — Deploy shell script recovery otomatis pada Connect REST API untuk menyalakan kembali task failed akibat gangguan jaringan transien tanpa intervensi manual.

← Sebelumnya: Scaling & Resource Allocation Berikutnya: Connector Anti-Pattern →