Handling Duplicate Message #

Ketika kita mengonfigurasi konsumen Apache Kafka menggunakan jaminan pemrosesan At-Least-Once (untuk memastikan tidak ada satu pun pesan bisnis yang hilang), konsekuensi logis yang tidak dapat dihindari adalah munculnya duplikasi pesan. Jaringan internet yang tidak stabil, kegagalan penulisan offset komit di tengah-tengah transaksi, atau proses rebalance dinamis adalah peristiwa biasa di kluster server terdistribusi. Menyerahkan urutan pemrosesan pesan sepenuhnya ke database hilir tanpa perlindungan khusus akan merusak konsistensi data bisnis kita (seperti mendebit saldo nasabah berkali-kali). Oleh karena itu, kita wajib membangun mekanisme Idempotent Consumer (Konsumen Idempoten) untuk menyaring pesan ganda sebelum memicu logika bisnis yang sensitif.

Mengapa Duplikasi Pesan Terjadi di Kafka? #

Sebelum merancang solusi, kita harus memahami dengan tepat bagaimana duplikasi pesan menyusup ke dalam sistem kita. Ada dua skenario klasik yang paling sering terjadi di lingkungan produksi:

Skenario 1: Kegagalan Pengiriman Komit Offset (Network Timeout) #

  1. Produser mengirimkan pesan ke Kafka broker dengan kunci unik transaksi TX-999 (Offset 500).
  2. Konsumen kita sukses menarik pesan tersebut melalui pemanggilan fungsi .poll().
  3. Konsumen memproses transaksi, memotong saldo, dan menulis record transaksi sukses ke database bisnis kita.
  4. Masalah: Ketika konsumen mengirimkan sinyal commitSync(Offset 500) kembali ke broker, koneksi jaringan terputus sementara atau broker coordinator mengalami jeda waktu. Broker tidak pernah menerima komit tersebut.
  5. Setelah koneksi pulih, konsumen (atau konsumen baru pasca-rebalance) mendapati committed offset terakhir masih berada di angka 499.
  6. Konsumen ditarik mundur untuk membaca kembali Offset 500 (TX-999). Jika database tidak memiliki filter pelindung, saldo nasabah akan terpotong untuk kedua kalinya.

Skenario 2: Rebalance di Tengah Pemrosesan Batch #

  1. Konsumen memanggil .poll() dan menerima 50 pesan sekaligus (Offset 1000 hingga 1049).
  2. Konsumen mulai memproses pesan satu per satu secara sinkron di thread utama.
  3. Setelah sukses memproses 30 pesan hingga Offset 1029, total waktu yang dihabiskan telah melampaui batas waktu max.poll.interval.ms (misalnya karena kueri database melambat).
  4. Group Coordinator mendeteksi keterlambatan pemanggilan .poll(), menganggap konsumen tersebut mati, dan memicu rebalance.
  5. Partisi dipindahkan ke konsumen lain. Konsumen baru membaca offset komit terakhir yang valid di broker, yaitu Offset 999 (karena komit batch sebelumnya baru akan dikirim setelah seluruh 50 pesan selesai diproses).
  6. Konsumen baru memproses ulang seluruh 50 pesan dari Offset 1000, padahal 30 pesan pertama sebenarnya sudah sukses ditulis ke database oleh konsumen sebelumnya.

Mekanisme Idempotent Consumer (Konsumen Idempoten) #

Mekanisme Idempotent Consumer (sering disebut Deduplication Pattern) adalah pola arsitektur di mana aplikasi konsumen dirancang untuk menyaring pesan duplikat secara mandiri sehingga eksekusi berulang dari pesan yang sama tidak akan mengubah status akhir sistem melebihi eksekusi pertama.

Ada dua pendekatan utama yang bisa kita terapkan untuk mencapai idempotensi ini:

1. Database Unique Constraint / UPSERT (Idempotensi Alami) #

Jika entitas data bisnis kita secara alami memiliki pengenal unik (Natural Unique Key) yang dikirimkan oleh produser (misalnya order_id, invoice_number, atau payment_id), kita bisa memanfaatkan fitur keamanan bawaan database RDBMS (seperti PostgreSQL, MySQL) atau NoSQL (seperti MongoDB).

  • Unique Constraints: Menandai kolom order_id di database sebagai UNIQUE PRIMARY KEY. Jika konsumen mencoba menulis data yang sama untuk kedua kalinya, database akan memotong transaksi dan melemparkan eksepsi error pelanggaran integritas data (Integrity Constraint Violation).
  • UPSERT (Insert or Update): Jika kita tidak ingin melemparkan error melainkan memperbarui data yang ada, kita dapat menggunakan perintah UPSERT:
    • PostgreSQL: INSERT ... ON CONFLICT (order_id) DO UPDATE SET ... atau DO NOTHING.
    • MySQL: INSERT INTO ... ON DUPLICATE KEY UPDATE ...
    • MongoDB: Menggunakan operasi db.collection.updateOne() dengan parameter upsert: true.

2. Dedicated Deduplication Table (Tabel Deduplikasi Khusus) #

Dalam banyak skenario nyata, logika bisnis kita tidak sesederhana menyimpan data ke satu tabel. Kita mungkin harus memicu pemanggilan API pihak ketiga, mengirim email notifikasi ke pengguna, atau menjalankan serangkaian kueri kompleks yang tidak memiliki primary key tunggal.

Untuk menangani skenario ini, kita harus membuat Tabel Deduplikasi Khusus (biasanya bernama processed_events atau event_deduplication) di dalam database transaksional kita.

  • Cara Kerja:
    1. Setiap kali produser mengirim pesan, ia wajib menyertakan ID UUID unik di dalam header atau payload pesan (disebut Event ID).
    2. Saat konsumen membaca pesan, ia membuka transaksi database baru (BEGIN TRANSACTION).
    3. Konsumen mencoba memasukkan Event ID tersebut ke dalam tabel processed_events.
    4. Jika proses insert sukses, konsumen melanjutkan eksekusi logika bisnis utama dan memperbarui database bisnis di dalam transaksi yang sama.
    5. Konsumen mengomit transaksi database (COMMIT TRANSACTION).
    6. Jika proses insert di langkah ke-3 gagal karena kunci duplikat, konsumen segera membatalkan transaksi (ROLLBACK), mengabaikan pesan tersebut, dan langsung melanjutkan ke pesan berikutnya.

Alur Kerja Deteksi Pesan Duplikat #

Diagram alir di bawah ini memvisualisasikan bagaimana alur keputusan yang harus dilalui oleh aplikasi konsumen kita saat menyaring pesan menggunakan tabel deduplikasi:

flowchart TD
    subgraph Alur_Pemrosesan
        Start["1. Konsumen Memanggil poll()"] --> GetRecord["2. Dapatkan Pesan (Event ID: EV-888)"]
        GetRecord --> StartTx["3. Buka Transaksi Database (BEGIN)"]
        StartTx --> TryInsert["4. Coba Insert EV-888 ke tabel processed_events"]
        
        TryInsert --> Success{"Apakah insert sukses?"}
        
        Success -- "Ya (Belum Pernah Diproses)" --> ProcessBiz["5. Eksekusi Logika Bisnis Utama"]
        ProcessBiz --> CommitTx["6. Komit Transaksi Database (COMMIT)"]
        CommitTx --> CommitOffset["7. Komit Offset ke Broker Kafka"]
        CommitOffset --> End["Selesai, Lanjut Pesan Berikutnya"]
        
        Success -- "Tidak (Duplikat Terdeteksi)" --> RollbackTx["8. Batalkan Transaksi Database (ROLLBACK)"]
        RollbackTx --> SkipMessage["9. Abaikan Pesan / Log Warning"]
        SkipMessage --> CommitOffset
    end

    style Success stroke:#fbc02d,stroke-width:2px
    style ProcessBiz stroke:#2e7d32,stroke-width:2px
    style SkipMessage stroke:#c62828,stroke-width:2px

Deduplikasi Berbasis Redis (In-Memory Distributed Cache) #

Pada aplikasi dengan throughput sangat tinggi (misalnya memproses ratusan ribu logistik log per detik), melakukan kueri INSERT ke database relasional (RDBMS) untuk setiap pesan dapat memicu kemacetan performa (disk I/O bottlenecks).

Sebagai alternatif, kita bisa menggunakan penyimpanan memori terdistribusi seperti Redis sebagai filter deduplikasi cepat sebelum data dikirim ke RDBMS.

  • Cara Kerja:
    • Kita memanfaatkan perintah atomik Redis SET key value EX seconds NX.
    • Parameter NX memastikan kunci hanya akan tersimpan jika belum ada di Redis.
    • Parameter EX memberikan batas waktu kedaluwarsa (Time-To-Live / TTL) secara otomatis agar memori Redis tidak membengkak tanpa batas.
    • Jika Redis mengembalikan status sukses, kita lanjutkan pemrosesan ke database. Jika gagal (kembalian null atau false), pesan segera diabaikan karena terbukti duplikat.
// CONTOH REDIS-BASED DEDUPLICATION DI JAVA (MENGGUNAKAN JEDIS)
String redisKey = "event:" + eventId;
// Simpan Event ID dengan TTL 24 jam (86400 detik) hanya jika key belum ada (NX)
String result = jedis.set(redisKey, "processed", new SetParams().nx().ex(86400));

if ("OK".equals(result)) {
    // ✓ Aman: Event ID belum pernah diproses sebelumnya
    executeDatabaseLogistics(record.value());
} else {
    // ✗ Duplikat terdeteksi di cache memori Redis
    log.warn("Mendeteksi pesan duplikat di Redis cache: {}", eventId);
}
  • Trade-Off: Pendekatan ini tidak sepenuhnya transaksional secara atomik (karena Redis dan RDBMS berada di sistem terpisah). Jika penulisan database gagal setelah Redis sukses di-set, kita harus memiliki mekanisme compensating transaction atau menghapus key di Redis agar pesan dapat di-retry kembali.

Hubungan dengan Transactional Outbox Pattern #

Untuk menjamin keandalan data end-to-end, penanganan duplikasi di sisi konsumen sering kali dipasangkan dengan Transactional Outbox Pattern di sisi produser.

  • Produser menulis data bisnis dan data event ke tabel outbox lokal dalam satu transaksi database lokal.
  • Komponen Debezium atau CDC (Change Data Capture) membaca tabel outbox dan mempublikasikannya ke Kafka broker secara asinkron.
  • Karena CDC menjamin pengiriman At-Least-Once ke broker, konsumen tetap wajib mengimplementasikan tabel deduplikasi.
  • Sinergi ini memastikan bahwa dari titik produser hingga konsumen, data dijamin tidak akan hilang dan status akhir sistem terjamin konsisten tanpa ada duplikasi transaksi.

Kode Penerapan: Implementasi Idempotent Consumer #

Mari kita bandingkan secara langsung kode implementasi yang rentan terhadap duplikasi dengan kode solusi yang kokoh menggunakan RDBMS deduplication.

Java SDK Anti-Pattern: Penyimpanan Data Tanpa Proteksi Idempotensi #

Kode di bawah ini menunjukkan kesalahan umum pengembang yang berasumsi bahwa database insert biasa sudah cukup aman karena Kafka berjalan lancar.

// ANTI-PATTERN: Menulis data tanpa penyaringan keunikan pesan
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
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) {
            // 解析 JSON payload
            Order order = parseJson(record.value());
            
            // ✗ JANGAN: Menggunakan query insert mentah tanpa unique constraint atau checking.
            // Jika terjadi rebalance atau network failure sebelum commitSync(),
            // fungsi ini akan membuat baris order baru di database dengan ID yang sama,
            // memicu duplikasi data di dashboard admin.
            saveOrderToDatabase(order.getId(), order.getCustomerId(), order.getAmount());
        }
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Java SDK Solusi: Menerapkan Deduplication Table Transaksional #

Kode di bawah ini menunjukkan solusi yang direkomendasikan dengan membungkus proses insert ke tabel deduplikasi dan penulisan bisnis dalam satu transaksi database lokal menggunakan JDBC.

// BENAR: Mengamankan pemrosesan menggunakan transaksi database transaksional dan constraint
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
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) {
            
            // Dapatkan Event ID unik dari header Kafka (atau dari JSON payload)
            String eventId = getHeaderValue(record.headers(), "event_id");
            Order order = parseJson(record.value());

            if (eventId == null) {
                log.error("Pesan tidak memiliki event_id unik, abaikan untuk keselamatan.");
                continue;
            }

            try (Connection conn = dataSource.getConnection()) {
                // ✓ 1. Mulai Transaksi Database Lokal
                conn.setAutoCommit(false);

                try {
                    // ✓ 2. Coba catat Event ID ke tabel deduplikasi processed_events.
                    // Kolom event_id adalah PRIMARY KEY.
                    try (PreparedStatement dedupStmt = conn.prepareStatement(
                        "INSERT INTO processed_events (event_id) VALUES (?)"
                    )) {
                        dedupStmt.setString(1, eventId);
                        dedupStmt.executeUpdate();
                    }

                    // ✓ 3. Jika berhasil, eksekusi logika bisnis utama pada transaksi yang sama
                    try (PreparedStatement bizStmt = conn.prepareStatement(
                        "INSERT INTO orders (id, customer_id, amount) VALUES (?, ?, ?)"
                    )) {
                        bizStmt.setString(1, order.getId());
                        bizStmt.setString(2, order.getCustomerId());
                        bizStmt.setDouble(3, order.getAmount());
                        bizStmt.executeUpdate();
                    }

                    // ✓ 4. Komit seluruh transaksi database secara utuh
                    conn.commit();
                    log.info("Sukses memproses event: {} dan menyimpan data bisnis.", eventId);

                } catch (SQLIntegrityConstraintViolationException e) {
                    // ✓ 5. DETEKSI DUPLIKAT: Terjadi pelanggaran primary key di langkah 2.
                    // Segera batalkan transaksi database agar tidak menulis data ganda.
                    conn.rollback();
                    log.warn("Pesan duplikat terdeteksi untuk Event ID: {}. Mengabaikan pesan dengan aman.", eventId);
                } catch (Exception ex) {
                    conn.rollback();
                    log.error("Terjadi error sistem, membatalkan transaksi.", ex);
                    throw ex; // Lemparkan eksepsi agar loop konsumen berhenti & retry dijalankan
                }
            } catch (SQLException sqle) {
                log.error("Kegagalan koneksi database", sqle);
            }
        }
        
        // ✓ 6. Selalu lakukan komit offset di akhir batch
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Ringkasan #

  • At-Least-Once Side-Effect — Penggunaan jaminan At-Least-Once di Apache Kafka memastikan data tidak pernah hilang, namun membawa risiko terjadinya duplikasi pesan akibat kegagalan komit offset.
  • Idempotent Consumer Pattern — Pola arsitektur wajib pada sisi konsumen untuk memastikan pemrosesan pesan yang sama secara berulang tidak merusak status akhir database bisnis.
  • Natural Unique Key — Pendekatan menggunakan kolom kunci unik bawaan dari data bisnis (seperti order_id) dan mengandalkan Unique Constraint RDBMS untuk memotong duplikasi data.
  • Deduplication Table — Tabel pembantu khusus (processed_events) untuk mencatat Event ID UUID unik sebelum mengeksekusi logika bisnis yang kompleks di dalam satu transaksi database lokal.
  • Non-Transactional Side-Effects — Operasi eksternal non-database (seperti kirim email atau potong saldo via pihak ketiga) harus diamankan dengan menyertakan Idempotency Key saat memanggil API luar.
  • Transaksional Atomik — Keamanan deduplikasi dijamin dengan memastikan pencatatan event ID dan eksepsi bisnis dibungkus dalam blok transaksi BEGIN ... COMMIT yang sama.

← Sebelumnya: At-Least-Once vs At-Most-Once   Berikutnya: Poison Message Strategy →

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