Poison Message Strategy #

Dalam pengolahan data terdistribusi secara real-time, tidak semua pesan yang masuk ke dalam antrean Kafka broker dilahirkan sempurna. Terkadang, kita akan menemui pesan yang memiliki format biner rusak (corrupt byte), skema yang tidak kompatibel dengan kode aplikasi kita, atau isi data yang melanggar logika aturan bisnis secara fatal (seperti nominal pembayaran minus). Pesan-pesan bermasalah ini dikenal sebagai Poison Message (Pesan Beracun). Jika aplikasi konsumen kita tidak dirancang secara khusus untuk menangani pesan beracun ini, thread utama poll loop akan mengalami kegagalan berulang (infinite crash loops), menghentikan pembacaan partisi, dan melumpuhkan seluruh pipeline transmisi data bisnis kita. Kita wajib menerapkan strategi penanganan pesan beracun yang elegan menggunakan pola Dead Letter Queue (DLQ), retry topic, dan mekanisme deteksi kegagalan yang aman.

Apa itu Poison Message dan Dampak Fatalnya? #

Definisi mendasar dari Poison Message adalah pesan di dalam topik Kafka yang secara konsisten memicu kegagalan pemrosesan oleh aplikasi konsumen kita, terlepas dari berapa kali konsumen mencoba memprosesnya kembali.

Pesan beracun biasanya dikelompokkan ke dalam kategori Fatal Error (Non-Transient Error), seperti:

  • Kegagalan Deserialisasi: Klien gagal mengubah biner byte menjadi objek (misalnya produser mengirim format XML sedangkan konsumen membaca format JSON).
  • Kerusakan Payload: Pesan terpotong di tengah jalan karena kegagalan jaringan saat produser menulis data.
  • Pelanggaran Aturan Bisnis: Data yang lolos validasi produser tetapi melanggar integritas database (misalnya nilai string terlalu panjang untuk kolom database).

Dampak Maut: Infinite Loop Blocking #

Jika kita menerapkan semantik At-Least-Once standar di mana offset baru dikomit setelah data selesai diproses, munculnya satu pesan beracun akan menghentikan seluruh sistem:

  1. Konsumen memanggil .poll() dan menarik pesan beracun (Offset 500).
  2. Kode aplikasi mencoba memproses pesan tersebut, mengalami error NullPointerException, dan melempar eksepsi.
  3. Karena eksepsi dilempar, perintah commit offset untuk 500 dilewati.
  4. Loop konsumen berputar kembali ke atas, memanggil .poll(), dan broker kembali memberikan pesan Offset 500 yang sama (karena belum dikomit).
  5. Aplikasi crash lagi. Siklus ini berulang terus-menerus (infinite retry loop). Akibatnya, pesan-pesan sehat di Offset 501 dan seterusnya tidak akan pernah diproses, memicu lonjakan metrik consumer lag yang luar biasa.

3 Strategi Penanganan Pesan Beracun #

Untuk menjaga kelangsungan operasional pipeline data kita, kita dapat memilih satu dari tiga strategi penanganan berikut berdasarkan tingkat toleransi audit data bisnis kita:

1. Tangkap dan Abaikan (Catch and Discard) #

Ini adalah pendekatan paling sederhana. Konsumen membungkus seluruh logika pemrosesan dengan blok try-catch.

  • Cara Kerja: Jika terjadi kegagalan pemrosesan akibat pesan beracun, konsumen menangkap error tersebut, menulis catatan log tingkat WARN atau ERROR, lalu tetap mengirimkan komit offset ke broker agar pointer maju ke pesan berikutnya.
  • Kelemahan: Pesan bermasalah dibuang begitu saja tanpa ada jejak audit biner asli. Kita tidak dapat merekonstruksi atau memproses ulang pesan tersebut secara manual setelah kode aplikasi diperbaiki.

2. Dead Letter Queue (DLQ) / Topik Pembuangan #

Dead Letter Queue (DLQ) adalah topik Kafka khusus yang ditugaskan untuk menampung seluruh pesan beracun yang gagal diproses. Ini adalah standar terbaik dalam arsitektur microservices modern.

  • Cara Kerja:
    1. Konsumen menarik pesan dari topik utama (misal: order-events).
    2. Saat pemrosesan gagal secara fatal, konsumen menangkap eksepsi tersebut.
    3. Konsumen bertindak sebagai produser sejenak: ia mempublikasikan pesan asli beserta metadata error (seperti stack trace error, nama konsumen, dan waktu kegagalan) ke topik DLQ (misal: order-events-dlq).
    4. Setelah pesan sukses terkirim ke DLQ, konsumen melakukan komit offset pada topik utama agar loop dapat melanjutkan membaca pesan sehat berikutnya.
flowchart TD
    subgraph Flow_DLQ
        Start["1. Ambil Pesan dari Topik order-events"] --> TryProcess["2. Jalankan Pemrosesan Data"]
        TryProcess --> Success{"Apakah sukses?"}
        
        Success -- "Ya" --> Commit["3. Komit Offset Utama ke Broker"]
        Success -- "Tidak (Crash)" --> CheckError{"Tipe Error?"}
        
        CheckError -- "Fatal (Poison)" --> PublishDLQ["4. Publikasikan ke order-events-dlq"]
        PublishDLQ --> Commit
        
        CheckError -- "Sementara (Transient)" --> PublishRetry["5. Kirim ke order-events-retry-5m"]
        PublishRetry --> Commit
        
        Commit --> End["Selesai, Siap Poll Berikutnya"]
    end
    
    style Success stroke:#fbc02d,stroke-width:2px
    style PublishDLQ stroke:#c62828,stroke-width:2px
    style PublishRetry stroke:#0288d1,stroke-width:2px

3. Retry Topics dengan Exponential Backoff #

Terkadang, kegagalan pemrosesan tidak disebabkan oleh pesan yang rusak secara fisik (poison), melainkan karena gangguan infrastruktur sementara (Transient Error), seperti database lokal overload sejenak atau API pihak ketiga sedang mengalami downtime.

  • Untuk transient error, langsung membuang pesan ke DLQ adalah kesalahan besar. Kita harus mencobanya kembali beberapa saat kemudian.
  • Solusi: Kita membuat topik retry berjenjang (misalnya order-events-retry-5m, order-events-retry-30m). Pesan dikirim ke topik retry tersebut dan dikonsumsi oleh kelompok konsumen retry yang dikonfigurasi untuk menunda pembacaan menggunakan jeda waktu (exponential backoff). Jika setelah beberapa kali percobaan tetap gagal, baru pesan dialihkan ke DLQ sebagai keputusan terakhir.

Solusi Spring Kafka: ErrorHandlingDeserializer #

Jika kita mengembangkan aplikasi menggunakan framework Spring Boot dengan modul Spring Kafka, ada satu keunikan arsitektur yang sering membingungkan pengembang: kegagalan deserialisasi terjadi sebelum pesan masuk ke kode listener/controller kita.

  • Masalah: Spring secara internal melakukan deserialisasi byte menjadi objek Java sebelum memanggil fungsi @KafkaListener. Jika deserialisasi ini melempar error, Spring listener container akan crash di luar blok try-catch yang kita tulis di dalam listener.
  • Solusi Spring: Spring menyediakan kelas pembungkus khusus bernama ErrorHandlingDeserializer.
    • Kelas ini membungkus deserializer asli (seperti JsonDeserializer).
    • Jika deserializer asli melempar error, ErrorHandlingDeserializer akan menangkap eksepsi tersebut, memblokir crash kontainer, lalu meneruskan objek error pembungkus khusus (FailedDeserializationInfo) ke listener kita.
    • Spring CommonErrorHandler (seperti DefaultErrorHandler yang dikombinasikan dengan DeadLetterPublishingRecoverer) secara otomatis akan mendeteksi objek error tersebut dan langsung mengalihkan pesan biner rusak tadi ke topik DLQ tanpa menyentuh kode bisnis kita sama sekali.

Alur Pemulihan dan Konsumsi Ulang dari DLQ (Re-drive Utility) #

Mengirim pesan beracun ke topik DLQ barulah langkah awal penyelamatan sistem. Setelah pesan berada di DLQ, apa yang harus kita lakukan? Pesan-pesan di DLQ tidak boleh dibiarkan menumpuk tanpa batas waktu.

Ada tiga langkah penanganan yang umum diterapkan di lingkungan produksi:

  1. Investigasi & Perbaikan Kode: Tim pengembang menganalisis stack trace error di header DLQ. Jika kegagalan disebabkan oleh bug logika (misalnya salah menangkap format timestamp), tim segera merilis hotfix perbaikan kode ke server produksi.
  2. Utilitas Re-drive (DLQ Re-drive Utility): Kita membuat aplikasi utilitas kecil mandiri yang bertugas mengonsumsi data dari topik DLQ, memperbaiki payload yang rusak (jika ada kesalahan ketik manual), lalu mempublikasikannya kembali ke topik utama (order-events) agar diproses ulang oleh kelompok konsumen utama yang sekarang sudah dipasang hotfix kode baru.
  3. Pembersihan Otomatis: Jika pesan di DLQ terbukti sebagai sampah palsu yang tidak bernilai bisnis, kita biarkan retention policy dari topik DLQ (misalnya 14 hari) menghapusnya secara otomatis dari disk broker.

Desain Arsitektur DLQ yang Ideal #

Ketika membangun sistem DLQ di Apache Kafka, kita disarankan untuk mengikuti aturan desain berikut:

  • Pertahankan Header Asli: Pesan yang dikirim ke DLQ harus mempertahankan key, payload biner asli, dan partisi asalnya.
  • Tambahkan Header Audit: Selalu sisipkan metadata error pada header pesan DLQ:
    • x-exception-message: Alasan singkat kegagalan (misalnya NullPointerException).
    • x-exception-stacktrace: Stack trace penuh dari kode Java kita.
    • x-original-topic & x-original-partition: Asal muasal topik dan partisi pesan.
    • x-original-offset: Nomor offset asli saat pesan gagal diproses.
  • Monitoring DLQ: Buatlah alert/alarm monitoring jika volume pesan di topik DLQ meningkat secara drastis, karena itu mengindikasikan adanya bug rilis kode baru di sistem produksi kita.

Kode Penerapan: Penanganan Pesan Beracun dan DLQ #

Mari kita bandingkan secara langsung kode implementasi konsumen yang rentan mengalami kemacetan infinite crash dengan kode solusi yang menerapkan pola DLQ secara aman.

Java SDK Anti-Pattern: Tanpa Penanganan Eksepsi (Infinite Crash Loop) #

Kode di bawah ini menunjukkan kesalahan fatal di mana konsumen akan langsung mandek selamanya jika bertemu dengan satu pesan beracun yang memicu error parsing.

// ANTI-PATTERN: Membiarkan runtime exception mematikan loop konsumen
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-parser-group");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // ✗ JANGAN: Melakukan parsing JSON tanpa try-catch penanganan error.
            // Jika record.value() berisi string rusak (misalnya XML mentah),
            // jsonParser akan melempar JsonParseException. Utas akan crash,
            // and offset tidak akan pernah dikomit. Konsumen akan membaca ulang
            // pesan beracun ini tiada henti pada panggilan poll() berikutnya.
            Order order = jsonParser.readValue(record.value(), Order.class);
            processOrder(order);
        }
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Java SDK Solusi: Menerapkan Dead Letter Queue (DLQ) #

Kode di bawah ini menunjukkan solusi yang direkomendasikan dengan menangkap eksepsi kegagalan secara spesifik, mengalirkan pesan rusak ke topik DLQ dengan metadata audit yang lengkap, dan mengomit offset agar sistem dapat melanjutkan pemrosesan.

// BENAR: Mengamankan pemrosesan menggunakan catch block dan pengalihan ke DLQ
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-parser-group");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order-events"));

// Inisialisasi produser internal untuk mengirim pesan ke DLQ
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
KafkaProducer<String, String> dlqProducer = new KafkaProducer<>(producerProps);

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            try {
                // Jalankan proses deserialisasi dan logika bisnis
                Order order = jsonParser.readValue(record.value(), Order.class);
                processOrder(order);
                
            } catch (JsonProcessingException | NullPointerException fatalException) {
                // ✓ 1. Tangkap error fatal yang masuk kategori Poison Message
                log.error("Pesan beracun terdeteksi pada offset: {}! Mengalihkan ke DLQ.", record.offset(), fatalException);
                
                // ✓ 2. Rakit pesan baru untuk dikirim ke topik DLQ (order-events-dlq)
                ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(
                    "order-events-dlq", 
                    record.key(), 
                    record.value()
                );
                
                // ✓ 3. Sisipkan metadata audit penuh ke dalam header pesan DLQ
                dlqRecord.headers().add("x-exception-message", fatalException.getMessage().getBytes());
                dlqRecord.headers().add("x-original-topic", record.topic().getBytes());
                dlqRecord.headers().add("x-original-partition", String.valueOf(record.partition()).getBytes());
                dlqRecord.headers().add("x-original-offset", String.valueOf(record.offset()).getBytes());
                
                try {
                    // Kirim pesan secara sinkron ke DLQ untuk menjamin pesan telah tersimpan sebelum commit
                    dlqProducer.send(dlqRecord).get();
                } catch (Exception producerEx) {
                    log.error("Gagal mengirimkan pesan beracun ke DLQ, hentikan loop!", producerEx);
                    throw new RuntimeException("DLQ Offline", producerEx);
                }
                
            } catch (TransientDatabaseException transientException) {
                // ✗ JANGAN kirim ke DLQ jika error bersifat sementara (transient)
                log.warn("Database sibuk sejenak, lemparkan error agar dicoba kembali (retry).");
                throw transientException;
            }
        }
        
        // ✓ 4. Selalu lakukan komit offset di akhir batch agar sistem terus maju
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} catch (Exception e) {
    log.error("Terjadi error fatal, menghentikan loop konsumen untuk rebalance.", e);
} finally {
    dlqProducer.close();
    consumer.close();
}

Ringkasan #

  • Poison Message (Pesan Beracun) — Pesan dengan format rusak, deserialisasi gagal, atau payload tidak kompatibel yang selalu memicu crash berulang pada konsumen.
  • Infinite Crash Loops — Hambatan fatal di mana konsumen terjebak membaca pesan rusak yang sama secara terus-menerus karena offset gagal dikomit akibat eksepsi error.
  • Dead Letter Queue (DLQ) — Topik Kafka cadangan khusus yang ditugaskan untuk menampung pesan bermasalah secara aman guna keperluan investigasi tim audit.
  • Audit Metadata Headers — Informasi penting seperti stack trace error, nama topik asal, nomor offset asli, dan waktu kegagalan wajib disematkan pada header pesan DLQ.
  • Transient vs Fatal Error — Hanya alihkan pesan ke DLQ jika error bersifat fatal/permanen (deserialisasi). Untuk transient error (database mati sejenak), terapkan mekanisme retry backoff.
  • ErrorHandlingDeserializer — Solusi framework Spring Kafka untuk menangkap error deserialisasi sebelum menyentuh listener utama, lalu secara otomatis mengalirkannya ke DLQ.
  • DLQ Re-drive Utility — Aplikasi bantu untuk mengonsumsi, memperbaiki, dan mengirimkan kembali pesan dari topik DLQ ke topik utama setelah bug aplikasi diperbaiki.
  • DLQ Alert Monitoring — Penumpukan pesan yang mendadak tinggi pada topik DLQ mengindikasikan adanya ketidakcocokan skema data pasca-rilis versi kode aplikasi baru.

← Sebelumnya: Duplicate Message   Berikutnya: At-Most-Once →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact