At-Least-Once #

Dalam dunia arsitektur sistem terdistribusi, kehilangan data merupakan skenario terburuk yang harus dihindari dengan segala cara oleh para arsitek dan pengembang aplikasi. Untuk menjamin bahwa setiap pesan yang diproduksi dan dikirimkan oleh sistem selalu sukses diproses di sistem tujuan, Apache Kafka menyediakan model jaminan pengiriman At-Least-Once (Setidaknya Sekali). Ini adalah tingkat jaminan default dan yang paling populer digunakan dalam ekosistem Kafka. Melalui semantik At-Least-Once, Kafka memberikan komitmen mutlak terhadap keandalan data (zero data loss). Setiap pesan dijamin akan berhasil dibaca dan diproses oleh sistem hilir setidaknya satu kali. Konsekuensi logis dari jaminan ini adalah munculnya potensi terjadinya pesan ganda atau duplikasi data di sisi konsumen jika terjadi kegagalan jaringan atau crash aplikasi tepat sebelum konfirmasi pembacaan (komit offset) tersimpan di broker.

Filosofi Utama At-Least-Once dan Konsep Zero Data Loss #

Filosofi utama di balik semantik At-Least-Once adalah keandalan mutlak lebih berharga daripada biaya penanganan data ganda. Di sistem monolitik tradisional, kita dapat menggunakan transaksi database lokal (ACID) untuk memastikan operasi penulisan data dan pembaharuan status berjalan secara atomik. Namun, di dalam arsitektur mikro (microservices) dan sistem terdistribusi skala besar, transaksi ACID lintas batas jaringan sangatlah lambat dan rentan terhadap kegagalan parsial (split-brain).

Kafka memecahkan masalah ini dengan menerapkan strategi penjaminan dua arah:

Sisi Produser: Produser tidak akan menganggap pengiriman pesan sukses sebelum ia menerima bukti tertulis (konfirmasi ACK) bahwa pesan tersebut telah disimpan dengan aman di beberapa broker replika. Jika ACK gagal diterima karena gangguan jaringan, produser akan terus mencoba mengirimkan ulang pesan tersebut.
Sisi Konsumen: Konsumen dilarang keras menginformasikan kepada broker bahwa ia telah selesai membaca suatu pesan sebelum logika bisnis aplikasi (seperti menulis ke database relasional, memanggil API pihak ketiga, atau memperbarui cache) dipastikan berhasil dijalankan tanpa ada eksepsi error. Pola ini dikenal dengan nama Process-First, Commit-Second (Proses Dulu, Komit Kemudian).

sequenceDiagram
    autonumber
    participant Prod as Produser Client
    participant Broker as Kafka Broker (ISR)
    participant Cons as Konsumen Client
    participant DB as Database Klien

    Note over Prod, Broker: "Sisi Produser: Jaminan Menyimpan (acks=all)"
    Prod->>Broker: ProduceRequest (Pesan X)
    Note over Broker: Broker menulis ke Leader & Follower (ISR)
    Broker-->>Prod: ACK Sukses (Pesan X aman di disk)

    Note over Broker, DB: "Sisi Konsumen: Process-First (At-Least-Once)"
    Cons->>Broker: poll() menarik Pesan X (Offset 500)
    Cons->>DB: Tulis Pesan X ke Database
    DB-->>Cons: Tulis Sukses
    Note over Cons: CRASH! (Server mati mendadak sebelum sempat mengirim komit offset)
    
    Note over Cons, Broker: "Proses Setelah Pemulihan (Recovery)"
    Note over Cons: Konsumen baru / restart membaca ulang
    Cons->>Broker: poll() dari Offset 500 (Broker tidak tahu crash sebelumnya)
    Cons->>DB: Tulis Pesan X ke Database (DUPLIKAT!)
    DB-->>Cons: Tulis Sukses
    Cons->>Broker: commitSync(Offset 500)
    Broker-->>Cons: Commit ACK (Offset 500 tersimpan aman)

Melalui diagram sekuensial di atas, kita dapat melihat dengan jelas bahwa pada langkah ke-6, data berhasil ditulis ke database eksternal klien. Namun, karena crash mendadak terjadi di langkah ke-7 sebelum komit offset terkirim, setelah konsumen pulih pada langkah ke-8, ia akan menarik kembali pesan yang sama (Offset 500) dari broker. Ini menyebabkan penulisan ulang data yang sama ke database pada langkah ke-10, menghasilkan duplikasi data.

Konfigurasi Produser untuk Jaminan Zero-Data-Loss #

Untuk memastikan bahwa produser tidak pernah kehilangan pesan sebelum pesan tersebut tersimpan di dalam kluster Kafka, kita harus mengonfigurasi parameter ketahanan (durability) produser sebagai berikut:

1. `acks=all` (atau `acks=-1`) #

Ini adalah parameter paling kritis untuk menjamin durabilitas.

Cara Kerja: Produser baru akan menganggap pengiriman pesan sukses setelah broker leader partisi berhasil mereplikasi pesan tersebut ke seluruh broker pengikut (followers) yang berada di dalam daftar In-Sync Replicas (ISR).
Sinergi dengan min.insync.replicas: Konfigurasi ini harus disandingkan dengan parameter broker min.insync.replicas (misalnya disetel ke 2 pada topik dengan Replication Factor = 3). Jika jumlah replika aktif yang sinkron kurang dari batas minimal ini, broker leader akan menolak menulis data baru dan memicu error NotEnoughReplicasException di sisi produser.

2. `retries=2147483647` (Integer.MAX_VALUE) #

Sejak Kafka 2.0, default nilai retries adalah nilai maksimal integer.

Cara Kerja: Jika terjadi gangguan jaringan sementara (misalnya jeda koneksi soket, GC pause pada broker leader, atau proses transisi pemilihan leader baru), produser akan secara aktif mencoba mengirim kembali pesan tersebut sampai berhasil.
Delivery Timeout: Batas waktu maksimal upaya retries ini dikontrol oleh properti delivery.timeout.ms (secara default disetel ke 120000 ms atau 2 menit). Selama timeout ini belum terlampaui, produser tidak akan menyerah mengirimkan ulang pesan.

3. `max.in.flight.requests.per.connection` #

Secara default, parameter ini bernilai 5. Jika terjadi retries pada salah satu batch request yang gagal sementara batch berikutnya di jaringan sudah sukses terkirim, urutan pesan di broker bisa menjadi acak (out-of-order).

Solusi Tanpa Idempotensi: Setel properti ini ke 1 (mengurangi throughput secara signifikan).
Solusi Modern: Gunakan enable.idempotence=true (sangat direkomendasikan karena menjaga urutan pesan tetap konsisten meskipun max.in.flight bernilai hingga 5).

Konfigurasi Konsumen: Pola Process-First #

Untuk menjamin semantik At-Least-Once di sisi konsumen, tanggung jawab komit offset berada sepenuhnya pada logika kode aplikasi kita. Kita harus menolak komit otomatis berbasis waktu yang tidak presisi.

Properti konsumen yang wajib disetel:

enable.auto.commit=false: Mengharuskan kita memicu pengiriman komit offset secara manual lewat kode.

Alur Kerja Konsumen yang Benar (Process-First) #

Klien memanggil .poll(Duration) untuk mengambil daftar record data dari broker.
Melakukan perulangan (looping) untuk memproses setiap record secara berurutan.
Menyimpan status transaksi atau hasil pemrosesan record ke database atau memanggil HTTP API eksternal.
Hanya setelah seluruh pemrosesan sukses tanpa ada eksepsi, panggil fungsi consumer.commitSync() atau consumer.commitAsync() untuk menginformasikan broker bahwa batch data tersebut telah selesai diproses.

flowchart TD
    Poll["Poll Data"] --> Process["Proses Logika Bisnis & Simpan ke DB"] --> Commit["Commit Offset ke Broker"]
    Process -- "Aplikasi Crash" --> ReRead["Data Dibaca Ulang dari Broker"]

Jika terjadi error saat memproses record ke-3 dari total 10 record yang ditarik, kode kita harus melempar eksepsi dan menghentikan proses sebelum memanggil fungsi komit. Dengan demikian, broker tetap mencatat offset lama. Saat konsumen baru mengambil alih partisi tersebut, ia akan memproses kembali 10 record dari awal, memastikan record ke-3 yang gagal tadi tidak terlewatkan.

Kronologi Detil Terjadinya Data Duplikat di Lingkungan Produksi #

Mari kita telusuri skenario nyata bagaimana pesan ganda dapat masuk ke database kita saat menggunakan semantik At-Least-Once.

Skenario 1: ACK Produser Hilang di Jaringan #

Produser mengirimkan pesan ProduceRequest(payload="Pesan A") ke Broker Leader.
Broker menulis pesan ke log disk lokal, menyalin ke ISR follower, dan membuat ACK sukses.
Tepat sebelum ACK dikirim ke kartu jaringan produser, switch jaringan di rak broker mengalami mati listrik sesaat. Paket data ACK gagal terkirim ke klien produser.
Klien produser menunggu hingga batas waktu request.timeout.ms (default 30 detik) terlampaui.
Karena tidak menerima ACK, produser menganggap pesan gagal terkirim akibat gangguan jaringan.
Produser melakukan retry otomatis dan mengirimkan kembali ProduceRequest(payload="Pesan A").
Broker menerima pesan tersebut. Karena broker menganggap ini adalah request baru yang sah (jika idempotensi dinonaktifkan), broker menulis “Pesan A” ke log disk partisi pada offset berikutnya (misalnya offset 101).
Broker mengirimkan ACK baru yang sukses. Produser menerimanya dengan baik.
Hasil: Di dalam topik Kafka sekarang terdapat dua pesan yang identik (“Pesan A”) di offset 100 dan offset 101. Keduanya akan dibaca oleh konsumen, menghasilkan duplikasi proses.

Skenario 2: Crash Konsumen saat Rebalance #

Konsumen C1 memanggil .poll() dan menerima 100 records dari partisi 2, mulai dari offset 1000 hingga 1099.
C1 sukses memproses 99 records (offset 1000 hingga 1098) dan menyimpannya ke database MySQL.
Tepat sebelum memproses record terakhir (offset 1099), broker mendeteksi bahwa konsumen baru C2 telah bergabung ke dalam grup. Group Coordinator memicu proses Rebalance.
Protokol rebalance lama (Eager Rebalance) memaksa C1 untuk melepaskan kepemilikan partisi 2 sebelum ia sempat memanggil commitSync().
Partisi 2 dialokasikan kembali ke konsumen baru C2.
C2 mulai membaca data dari partisi 2. Karena offset terakhir yang dikomit di broker masih bernilai 1000, C2 akan menarik kembali records mulai dari offset 1000 hingga 1099.
Hasil: 99 records yang sudah sukses diproses oleh C1 akan ditulis ulang ke database MySQL oleh C2, memicu duplikasi data sebanyak 99 baris jika tidak ada penyaringan idempoten di sisi database.

Taktik Penyelamatan: Bagaimana Menghadapi Duplikasi Pesan? #

Karena duplikasi data adalah konsekuensi logis yang tidak dapat dihindari pada semantik At-Least-Once, aplikasi konsumen kita wajib memiliki pertahanan untuk menyaring pesan ganda tersebut. Berikut adalah tiga taktik penyelamatan yang umum digunakan:

1. Desain Logika Bisnis Idempoten (Idempotency) #

Operasi idempoten adalah operasi yang jika dijalankan berulang kali dengan input yang sama akan memberikan hasil akhir yang sama tanpa mengubah status sistem lebih dari sekali.

Contoh Non-Idempoten: UPDATE users SET balance = balance + 10000 WHERE id = 1. Jika dijalankan dua kali karena duplikasi pesan, saldo pengguna bertambah 20.000 (salah!).
Contoh Idempoten: UPDATE users SET balance = 50000 WHERE id = 1. Berapa kali pun dijalankan, saldo akhir tetap 50.000.

2. Validasi Unique Constraints (Kunci Unik Database) #

Memanfaatkan fitur kunci unik (unique key/primary key constraint) pada database relasional (SQL) atau database dokumen (NoSQL).

Cara Kerja: Sebelum menulis data transaksi, buat kolom gabungan unik (misalnya gabungan transaction_id dan event_timestamp) sebagai Primary Key.
Penanganan: Jika pesan duplikat datang, database akan melempar eksepsi pelanggaran kunci unik (UniqueConstraintViolationException). Kode konsumen kita cukup menangkap eksepsi ini, mencatat log peringatan, dan secara aman mengabaikannya (melanjutkan ke langkah commit offset).

3. Tabel Deduplikasi Menggunakan Redis (Idempotent Consumer Pattern) #

Jika database tujuan kita tidak mendukung unique constraints secara alami (seperti menulis ke Elasticsearch atau memanggil API SMS pihak ketiga), kita dapat menggunakan Redis sebagai filter deduplikasi cepat.

Cara Kerja: Setiap kali menerima pesan, konsumen memeriksa apakah message_id atau event_uuid sudah ada di Redis menggunakan operasi atomik SETNX (Set if Not Exists) dengan batas kedaluwarsa waktu (Time to Live / TTL).
Logika: Jika SETNX mengembalikan nilai true, proses pesan dan komit. Jika mengembalikan false, buang pesan tersebut secara senyap karena telah diproses sebelumnya oleh thread lain.

Kode Penerapan Java SDK: Manual Commit Terbaik #

Berikut adalah contoh implementasi konsumen Kafka menggunakan Java SDK yang menerapkan semantik At-Least-Once melalui penanganan commit manual, penanganan error jaringan, dan penutupan koneksi secara bersih (graceful shutdown).

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class AdvancedAtLeastOnceConsumer {
    private static final Logger log = LoggerFactory.getLogger(AdvancedAtLeastOnceConsumer.class);
    private static final AtomicBoolean running = new AtomicBoolean(true);
    private static KafkaConsumer<String, String> consumer;

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // =====================================================================
        // KONFIGURASI ZERO DATA LOSS (AT-LEAST-ONCE)
        // =====================================================================
        // ✓ Wajib matikan auto-commit agar kita bisa memastikan pemrosesan sukses dulu
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        
        // ✓ Jika offset tidak ada di broker, mulai membaca dari posisi terlama
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("customer-orders"));

        // Setup Shutdown Hook untuk penutupan koneksi secara bersih
        final Thread mainThread = Thread.currentThread();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("Mendeteksi sinyal terminasi, menghentikan poll loop...");
            running.set(false);
            // Memicu WakeupException pada utas utama yang sedang tertahan di .poll()
            consumer.wakeup();
            try {
                mainThread.join();
            } catch (InterruptedException e) {
                log.error("Gagal menunggu thread utama selesai", e);
            }
        }));

        try {
            while (running.get()) {
                // Tarik batch pesan dari broker
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                if (!records.isEmpty()) {
                    log.info("Menerima {} records dari poll.", records.count());
                    boolean batchSuccess = true;

                    for (ConsumerRecord<String, String> record : records) {
                        try {
                            // ✓ PROSES DULU: Jalankan logika bisnis penting (Simpan ke DB / hitung transaksi)
                            processOrderRecord(record);
                        } catch (Exception e) {
                            // ✗ JANGAN: Lanjut memproses record berikutnya jika terjadi fatal error database.
                            // Kita harus menandai batch gagal agar offset tidak dikomit ke broker.
                            log.error("✗ Fatal error saat memproses offset {}. Menghentikan batch.", record.offset(), e);
                            batchSuccess = false;
                            break;
                        }
                    }

                    // ✓ KOMIT KEMUDIAN: Hanya kirim komit offset jika SELURUH records sukses diproses
                    if (batchSuccess) {
                        try {
                            // Menggunakan commitSync() secara sinkron untuk jaminan pemrosesan block yang kuat.
                            // commitSync() akan memblokir thread sampai broker membalas dengan sukses.
                            consumer.commitSync();
                            log.info("✓ Sukses mengomit offset batch saat ini.");
                        } catch (CommitFailedException e) {
                            log.error("✗ Gagal mengomit offset ke broker (Rebalance sedang terjadi atau timeout):", e);
                            // Kegagalan di sini berarti data akan diproses ulang pada poll berikutnya (aman dari data loss).
                        }
                    } else {
                        // Skenario Penyelamatan: Jika pemrosesan batch gagal, kita harus memutar balik offset
                        // pembacaan ke posisi offset terendah dari batch yang gagal agar tidak terjadi loncatan offset.
                        seekToFirstUncommittedOffset(records);
                    }
                }
            }
        } catch (WakeupException e) {
            // Abaikan eksepsi ini karena dipicu secara sengaja saat shutdown aplikasi
            log.info("Poll loop dihentikan secara bersih via Wakeup API.");
        } catch (Exception e) {
            log.error("Terjadi eksepsi tidak terduga pada consumer loop", e);
        } finally {
            try {
                // Selalu lakukan komit sinkron terakhir sebelum menutup koneksi
                consumer.commitSync();
            } catch (Exception e) {
                log.warn("Gagal melakukan komit final saat penutupan konsumen.");
            } finally {
                consumer.close();
                log.info("Koneksi konsumen ditutup secara bersih.");
            }
        }
    }

    private static void processOrderRecord(ConsumerRecord<String, String> record) throws Exception {
        log.info("Memproses pesanan: Key={}, Offset={}, Value={}", record.key(), record.offset(), record.value());
        // Simulasi error database acak untuk pengujian ketahanan data
        if (record.value().contains("INVALID_PAYMENT")) {
            throw new Exception("Koneksi database terputus atau format pembayaran tidak valid.");
        }
        // Logika bisnis sukses
    }

    private static void seekToFirstUncommittedOffset(ConsumerRecords<String, String> records) {
        // Logika memutar balik pointer baca konsumen ke offset terendah dari batch yang gagal
        records.partitions().forEach(partition -> {
            long minOffset = records.records(partition).get(0).offset();
            log.warn("Memutar balik pointer baca partisi {} ke offset {}", partition, minOffset);
            consumer.seek(partition, minOffset);
        });
    }
}

Perbandingan Jaminan Pengiriman Data #

Karakteristik	At-Most-Once	At-Least-Once	Exactly-Once
Risiko Data Loss	Sangat Tinggi	Nol Besar (Zero Loss)	Nol Besar (Zero Loss)
Risiko Duplikasi	Nol (No Duplicates)	Tinggi (Potensi Duplikat)	Nol (No Duplicates)
Throughput & Latensi	Paling Maksimal	Sedang-Tinggi	Rendah-Sedang (Overhead Transaksi)
Urutan Operasi Klien	Commit-First	Process-First	Atomik Transaksional
Kebutuhan Deduplikasi	Tidak Butuh	Sangat Wajib	Tidak Butuh (Kafka-to-Kafka)

Ringkasan #

At-Least-Once (Setidaknya Sekali) — Jaminan pengiriman di mana setiap pesan dipastikan sukses diproses minimal satu kali oleh konsumen, memprioritaskan keamanan data di atas risiko duplikasi.

Process-First, Commit-Second — Strategi penulisan kode di mana konsumen memproses seluruh logika bisnis dan penyimpanan database terlebih dahulu sebelum mengirimkan sinyal komit offset ke broker.

Durabilitas Sisi Produser — Dicapai dengan mengonfigurasi acks=all dan retries maksimal, memastikan pesan telah aman disalin ke daftar anggota In-Sync Replicas (ISR) sebelum dianggap sukses.

Pemicu Duplikasi Data — Disebabkan oleh hilangnya paket ACK di jaringan saat pengiriman produser, atau crash aplikasi konsumen di tengah jalan pasca-penulisan database namun pra-commit offset.

Taktik Deduplikasi Database — Penggunaan unique/primary key constraints pada database SQL untuk menangkap eksepsi pelanggaran kunci unik secara otomatis saat pesan ganda tiba.

Deduplikasi In-Memory Redis — Implementasi filter caching cepat menggunakan perintah atomik SETNX untuk menyaring ID pesan yang telah selesai dieksekusi sebelumnya oleh thread lain.

Graceful Shutdown Hook — Penerapan API consumer.wakeup() untuk menginterupsi utas konsumen yang sedang memblokir jaringan, mencegah terjadinya rebalance liar saat aplikasi dimatikan.

← Sebelumnya: At-Most-Once Berikutnya: Exactly-Once →