Exactly-Once Processing #

Ketika kita merancang sistem pemrosesan data real-time untuk kasus penggunaan yang kritis — seperti pemrosesan transaksi keuangan, penghitungan inventaris logistik, atau sistem billing berbasis penggunaan — toleransi terhadap kesalahan data adalah nol. Kehilangan pesan (data loss) tidak dapat diterima, dan duplikasi pesan (data duplication) akibat kegagalan jaringan atau rebalance kluster juga sangat berbahaya. Untuk memecahkan tantangan ini secara otomatis, Apache Kafka Streams menyediakan jaminan pemrosesan Exactly-Once Processing (EOS - Semantik Tepat Sekali). Fitur ini memastikan bahwa setiap record masukan diproses tepat satu kali, bahkan jika terjadi kegagalan server di tengah jalan. Melalui artikel ini, kita akan membedah arsitektur internal EOS di Kafka Streams, cara kerja transaksi atomik terintegrasi dengan Two-Phase Commit (2PC), perbedaan EOS v1 vs EOS v2, konfigurasi kode Java yang aman, implikasi performa, batasan EOS, serta checklist kesiapan produksi.

Bagaimana Exactly-Once Berkerja di Kafka Streams #

Jaminan Exactly-Once Processing di Kafka Streams dicapai melalui integrasi yang sangat erat antara tiga komponen dasar Apache Kafka: Idempotent Producer, Transaction Coordinator, dan Consumer Isolation Level.

Di dalam Kafka Streams, alur kerja pemrosesan data selalu mengikuti pola Read-Process-Write (Baca-Proses-Tulis):

Read: Konsumen internal membaca record dari topik input.
Process: Logika bisnis mengeksekusi record tersebut, memperbarui state store lokal (RocksDB), dan bersiap membackup status ke changelog.
Write: Produser internal mengirimkan record hasil transformasi ke topik output dan menuliskan perubahan status ke topik changelog.

Protokol Two-Phase Commit (2PC) di Kafka #

Untuk menyatukan ketiga langkah di atas menjadi satu transaksi yang atomik (All-or-Nothing), Kafka Streams menggunakan variasi protokol terdistribusi klasik Two-Phase Commit (2PC):

Tahap 1: Prepare Commit (Persiapan) Ketika interval commit terlampaui (diatur oleh commit.interval.ms), thread Kafka Streams mengirimkan permintaan penutupan transaksi ke Transaction Coordinator. Coordinator menuliskan catatan khusus bertuliskan PrepareCommit ke topik internal log transaksi __transaction_state. Pada tahap ini, status transaksi dikunci secara permanen di broker. Jika coordinator crash setelah titik ini, coordinator pengganti yang aktif akan melanjutkan komit transaksi secara otomatis.
Tahap 2: Commit (Penyelesaian / Commit Markers) Setelah status PrepareCommit tertulis dengan aman di log, Transaction Coordinator mengirimkan instruksi WriteTxnMarkerRequest ke semua broker leader dari partisi topik output dan topik changelog yang terlibat. Broker-broker tersebut akan menuliskan record kontrol khusus bernama Commit Marker (atau Abort Marker jika dibatalkan) di log partisi fisik disk. Setelah semua marker terkirim, coordinator menandai transaksi selesai dengan menulis status CompleteCommit ke topik status transaksi.

sequenceDiagram
    autonumber
    participant Cons as Konsumen KStreams (Read)
    participant TC as Transaction Coordinator
    participant Out as Topik Output & Changelog (Write)
    participant Offset as Topik __consumer_offsets (Commit)

    Note over Cons, TC: "Siklus Transaksi Dimulai"
    Cons->>TC: Mulai Transaksi (beginTransaction)
    Cons->>Out: Kirim data hasil olahan (Produce)
    Cons->>Out: Kirim backup update RocksDB (Changelog)
    
    Note over Cons, Offset: "Mengunci Offset Konsumsi"
    Cons->>TC: Daftarkan offset konsumen (sendOffsetsToTransaction)
    TC->>Offset: Tulis status offset sementara (Uncommitted)

    Note over Cons, TC: "Fase Komit Atomik (2-Phase Commit)"
    Cons->>TC: Commit Transaksi (commitTransaction)
    TC->>Out: Tulis Commit Marker ke Log
    TC->>Offset: Ubah status Offset menjadi Committed
    TC-->>Cons: Transaksi Selesai & Sukses!

Evolusi EOS: EOS v1 (Alpha) vs EOS v2 (Exactly-Once Beta/Production) #

Apache Kafka memperkenalkan jaminan EOS pertama kali pada versi 0.11 (dikenal sebagai EOS v1), dan menyempurnakannya secara dramatis pada versi 2.5 melalui pengenalan EOS v2 (diaktifkan lewat konfigurasi exactly_once_v2).

1. EOS v1 (Menggunakan transactional.id per Task) #

Pada EOS v1, Kafka Streams membuat satu instansi produser transaksional terpisah untuk setiap StreamTask yang berjalan di dalam aplikasi.

Overhead Tinggi: Jika aplikasi kita memproses 50 partisi, akan ada 50 instansi produser transaksional aktif. Masing-masing produser harus berinteraksi dengan Transaction Coordinator secara mandiri, memicu pembuatan koneksi soket TCP yang melimpah dan membebani memori JVM.
Latensi Rebalance: Setiap kali terjadi rebalance, koordinasi penutupan transaksi lama untuk puluhan task memakan waktu yang sangat lama, memperlambat proses pemulihan kluster.

2. EOS v2 (Menggunakan transactional.id per StreamThread) #

EOS v2 didesain untuk menyingkirkan hambatan skalabilitas di atas dengan memetakan transaksi di tingkat StreamThread, bukan lagi tingkat StreamTask.

Satu Produser per Thread: Jika satu instansi aplikasi kita memiliki num.stream.threads=2, maka hanya ada 2 instansi produser transaksional yang dibuat, tidak peduli berapa ratus task yang dikelola oleh thread-thread tersebut.
Skalabilitas Luar Biasa: Mengurangi beban koneksi TCP ke broker hingga lebih dari 90%, menurunkan penggunaan CPU broker, dan mempercepat proses rebalancing kluster secara masif.
Sejak Kafka 3.0, EOS v1 telah ditandai sebagai deprecated (usang). Kita wajib menggunakan EOS v2 untuk semua aplikasi baru di lingkungan produksi.

Batasan Jaminan EOS di Kafka Streams #

Penting bagi kita untuk menyadari batas-batas teoretis di mana jaminan Exactly-Once ini berlaku. EOS di Kafka bukanlah solusi ajaib yang menyelesaikan semua masalah konsistensi sistem terdistribusi:

Hanya Berlaku untuk Aliran Kafka-to-Kafka: Jaminan Exactly-Once hanya dijamin penuh jika data dibaca dari topik Kafka dan ditulis kembali ke topik Kafka. Jika topologi Kafka Streams kita melakukan penulisan ke sistem luar (seperti melakukan HTTP POST call ke REST API eksternal atau query INSERT ke PostgreSQL) di tengah-tengah pemrosesan, transaksi Kafka tidak dapat melakukan rollback pada database eksternal tersebut jika transaksi dibatalkan di broker.
Mitigasi Sistem Eksternal: Untuk interaksi dengan sistem luar, kita harus menerapkan pola Idempotent Consumer (misal menggunakan constraint unique key di database) atau menggunakan pola Transactional Outbox Pattern dikombinasikan dengan Kafka Connect.

Kode Penerapan: Anti-Pattern vs Solusi Exactly-Once Terkelola #

Mari kita bandingkan pendekatan berbahaya yang mencoba menangani duplikasi secara manual dengan implementasi EOS resmi di Kafka Streams.

Kasus Penggunaan #

Kita memproses data saldo transaksi pembayaran (account-transactions). Kita ingin memastikan bahwa penambahan saldo berjalan untuk setiap akun diproses tepat satu kali tanpa ada duplikasi akibat rebalance kluster.

Anti-Pattern: Mengelola Deteksi Duplikat Manual dengan Redis Lock #

Mencoba membuat database eksternal (seperti Redis) sebagai tempat penyimpanan kunci pengunci transaksional (distributed lock) untuk menyaring record duplikat secara manual adalah anti-pattern yang merusak performa.

// ANTI-PATTERN: Membangun mekanisme idempotensi manual menggunakan Redis Distributed Lock.
// ✗ Sangat lambat (menambah overhead latensi jaringan), rawan deadlock, dan rentan terhadap kebocoran lock.
public class VolatileIdempotencyProcessor {
    private static final JedisPool jedisPool = new JedisPool("localhost", 6379);

    public static void build(StreamsBuilder builder) {
        builder.<String, String>stream("account-transactions")
            .filter((userId, transactionJson) -> {
                String txId = extractTxId(transactionJson);
                try (Jedis jedis = jedisPool.getResource()) {
                    // ✗ SANGAT BURUK: Blocking write lock ke Redis per event!
                    Long isNew = jedis.setnx("lock:" + txId, "locked");
                    if (isNew == 1) {
                        // Kunci disetel selama 10 menit pengaman
                        jedis.expire("lock:" + txId, 600);
                        return true; // Loloskan untuk diproses
                    }
                    return false; // Duplikat terdeteksi di level Redis, buang event
                }
            })
            .to("processed-transactions");
    }

    private static String extractTxId(String json) { return "tx_123"; }
}

Solusi Praktis: Mengaktifkan EOS v2 secara Terintegrasi #

Di bawah ini adalah cara yang benar dan direkomendasikan secara resmi. Kita mengaktifkan jaminan Exactly-Once dengan menyetel parameter processing.guarantee di level konfigurasi. Kafka Streams akan mengoordinasikan transaksi atomik RocksDB changelog dan commit offset secara transparan di memori lokal tanpa butuh infrastruktur eksternal tambahan.

// BENAR: Mengaktifkan jaminan pemrosesan Exactly-Once (EOS v2) di Kafka Streams.
// ✓ Aman dari kehilangan data, ter-scale secara otomatis, latensi mikrodetik.
public class ResilientExactlyOnceApp {
    
    public static Properties createConfiguration() {
        Properties config = new Properties();
        
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "finance-accounting-service");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // =====================================================================
        // KONFIGURASI PENTING EXACTLY-ONCE
        // =====================================================================
        // ✓ Mengaktifkan jaminan Exactly-Once v2 (EOS v2) secara global
        // Secara otomatis mengaktifkan parameter acks=all di produser internal
        // dan memaksa isolasi level read_committed di konsumen internal
        config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

        // ✓ Sinergi Interval Commit
        // Pada EOS, commit interval default diturunkan dari 30 detik menjadi 100 milidetik
        // untuk memastikan transaksi segera dikomit dan meminimalkan lag konsumen hilir
        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

        // Menyetel batas timeout transaksi internal menjadi 1 menit untuk mitigasi LSO Blockage
        config.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60000);

        return config;
    }

    public static void main(String[] args) {
        Properties config = createConfiguration();
        StreamsBuilder builder = new StreamsBuilder();
        
        // Membaca aliran transaksi keuangan
        KStream<String, String> transactions = builder.stream(
            "account-transactions",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // Melakukan akumulasi saldo berjalan secara stateful
        // RocksDB changelog dan pengiriman data output dijamin atomik
        transactions
            .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
            .aggregate(
                () -> 0.0,
                (userId, txJson, currentBalance) -> currentBalance + parseAmount(txJson),
                Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("account-balance-store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Double())
            )
            .toStream()
            .to("account-balance-alerts", Produced.with(Serdes.String(), Serdes.Double()));

        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, config);
        
        streams.start();
    }

    private static double parseAmount(String json) { return 150000.0; }
}

Implikasi Performa dan Bahaya LSO Blockage #

Meskipun EOS v2 memberikan kemudahan luar biasa dalam menjaga konsistensi data, kita harus mengantisipasi dua konsekuensi performa utama saat mengaktifkannya di lingkungan produksi:

1. Penurunan Throughput Maksimal #

Overhead Marker: Setiap transaksi yang dikomit menuliskan satu record tambahan berukuran kecil bernama Commit/Abort Marker ke log fisik partisi broker.
Dampak: Menyalakan EOS v2 umumnya akan menurunkan throughput maksimum penulisan sebesar 10% - 15% karena beban tambahan untuk Two-Phase Commit (2PC) internal.

2. Bahaya LSO Blockage (Last Stable Offset Menggantung) #

Konsumen downstream yang membaca hasil olahan Kafka Streams dengan isolasi read_committed tidak dapat membaca record apa pun melewati batas Last Stable Offset (LSO). LSO adalah offset dari record transaksi aktif tertua yang belum dikomit.

Skenario Masalah: Jika salah satu instance Kafka Streams mengalami pembekuan (GC Pause panjang atau hang) di tengah transaksi aktif, transaksi tersebut tidak akan pernah memanggil commitTransaction() atau abortTransaction() sebelum timeout terlampaui.
Dampak: Batas LSO di partisi broker akan tertahan di posisi lama. Akibatnya, konsumen hilir (read_committed) akan berhenti membaca (stuck) dan mengalami lonjakan lag yang parah, meskipun aplikasi produser lain terus mengirimkan ribuan pesan baru ke partisi yang sama. Konsumen hilir baru akan lancar kembali setelah transaksi yang menggantung tersebut dibatalkan otomatis setelah durasi transaction.timeout.ms terlampaui.

Checklist Kesiapan Produksi Exactly-Once (EOS v2) #

Sebelum kita merilis aplikasi Kafka Streams dengan EOS v2 ke lingkungan produksi, kita wajib memverifikasi bahwa parameter kluster broker dan klien telah dikonfigurasi dengan aman:

1. Konfigurasi Tingkat Broker (Broker Configuration) #

transaction.state.log.replication.factor=3: Memastikan replikasi log transaksi internal kluster Kafka aman dari kematian broker tunggal.
transaction.state.log.min.isr=2: Memastikan status transaksi baru berhasil ditulis hanya jika ISR minimal berjumlah dua broker.
transaction.id.expiration.ms=604800000 (7 hari): Batas waktu penyimpanan ID transaksi sebelum dihapus secara otomatis dari broker.

2. Konfigurasi Tingkat Klien (KStreams Client Configuration) #

processing.guarantee=exactly_once_v2: Parameter wajib untuk menyalakan EOS v2 terkompresi.
commit.interval.ms=100 (atau maksimal 500ms): Menjamin latensi pengiriman status transaksi tetap pendek.
transaction.timeout.ms=60000 (1 menit): Batas waktu pencegahan LSO blockage agar sistem hilir tidak mengalami hang terlalu lama saat crash.

Ringkasan #

Exactly-Once Processing (EOS) — Jaminan pemrosesan di mana setiap event masukan diproses tepat satu kali tanpa kehilangan atau duplikasi data meskipun terjadi kegagalan.

Read-Process-Write Atomicity — Protokol pengikatan offset konsumen, penulisan state store changelog, dan output data ke dalam satu transaksi atomik terdistribusi.

processing.guarantee=“exactly_once_v2” — Konfigurasi standar industri sejak Kafka 2.5 untuk menyalakan EOS v2 yang menggunakan satu produser per StreamThread.

read_committed — Tingkat isolasi konsumen hilir yang memastikan konsumen membuang pesan transaksi yang gagal (aborted) dan hanya membaca pesan sukses (committed).

Last Stable Offset (LSO) — Batas offset terujung di broker yang menunjukkan pesan transaksi aktif tertua yang belum ditutup, bertindak sebagai filter pengaman bagi konsumen.

LSO Blockage — Fenomena kemacetan konsumsi hilir akibat adanya transaksi menggantung di broker yang menahan pergeseran batas LSO melewati batas normal.

← Sebelumnya: Join Stream and Table Berikutnya: Plan vs SSL vs SASL →