Exactly-Once Semantics #

Dalam arsitektur pemrosesan data real-time, keakuratan data adalah kunci utama. Saat kita menggunakan model pengiriman pesan asinkron, kegagalan infrastruktur di tengah jalan sering kali memaksa sistem kita melakukan pemrosesan ulang data yang memicu duplikasi data (model At-Least-Once) atau bahkan kehilangan data (model At-Most-Once). Untuk mencapai tingkat keandalan tertinggi, Apache Kafka memperkenalkan jaminan Exactly-Once Semantics (EOS). EOS bukan sekadar fitur tunggal, melainkan sebuah integrasi arsitektural yang rumit antara fitur Idempotent Producer dengan Transaction API di sisi produser, berkolaborasi dengan penyetelan tingkat isolasi konsumen (isolation level). Jaminan EOS memastikan bahwa di dalam alur pemrosesan data, setiap pesan masukan diproses, status internal diperbarui, dan pesan hasil keluaran ditulis ke topik tujuan secara atomik — semuanya terjadi tepat satu kali, tidak kurang dan tidak lebih, meskipun kluster broker atau aplikasi kita mengalami crash di tengah proses.

Konsep Dasar Exactly-Once Semantics (EOS) #

Untuk memahami mengapa EOS sangat revolusioner, kita harus melihat konteks umum di mana jaminan ini paling dibutuhkan: pola pemrosesan read-process-write (baca, proses, dan tulis).

Pola ini merupakan fondasi dari arsitektur stream processing (seperti Kafka Streams atau aplikasi microservices kustom):

Read: Konsumen membaca pesan dari Topik Input A (misal: PaymentRequests).
Process: Aplikasi memproses data tersebut (misal: memvalidasi saldo dan memotong biaya admin).
Write: Produser menulis pesan hasil kalkulasi ke Topik Output B (misal: ProcessedTransactions) dan secara bersamaan mencatat offset pesan input ke topik internal __consumer_offsets.

Tanpa adanya sistem transaksional (EOS), jika aplikasi kita mengalami crash tepat setelah langkah 3 (menulis ke Topik Output B) namun sebelum sempat memperbarui offset di langkah 4, maka saat aplikasi dinyalakan ulang, ia akan membaca kembali pesan input yang sama dan mengirimkan pesan transaksi ganda ke Topik Output B.

EOS menyelesaikan masalah ini dengan memperlakukan langkah Write (Output) dan Write (Offsets) sebagai satu kesatuan transaksi atomik. Jika salah satu langkah gagal, seluruh transaksi dibatalkan (aborted), dan konsumen hilir tidak akan pernah melihat data yang setengah matang tersebut.

Arsitektur Transaksional Kafka: Peran Transaction Coordinator #

Sistem transaksi terdistribusi Kafka tidak mengandalkan protokol two-phase commit (2PC) tradisional yang lambat dan memblokir koneksi database. Kafka menerapkan arsitektur transaksional berbasis log yang dikoordinasikan oleh komponen internal bernama Transaction Coordinator.

Berikut adalah tiga komponen arsitektur utama yang mengelola transaksi Kafka:

1. Transaction Coordinator #

Ini adalah broker Kafka khusus yang bertugas mengelola siklus hidup transaksi dari produser. Tugasnya mirip dengan Group Coordinator yang mengelola konsumen. Broker leader dari topik internal __transaction_state otomatis bertindak sebagai Transaction Coordinator untuk produser terkait.

2. Topik Jurnal Transaksi (`__transaction_state`) #

Seluruh perubahan status transaksi (Ongoing, PrepareCommit, CompleteCommit) dicatat secara permanen di dalam topik internal berpartisipasi tinggi bernama __transaction_state. Topik ini direplikasi ke seluruh broker kluster untuk menjamin bahwa jika Transaction Coordinator aktif mati, broker ISR pengganti dapat membaca jurnal tersebut dan melanjutkan koordinasi transaksi yang menggantung.

3. Properti `transactional.id` #

Agar transaksi dapat bertahan dari skenario crash dan restart produser, kita wajib mendefinisikan properti transactional.id yang unik dan persisten di sisi produser (misal: transactional.id=produser-transaksi-pembayaran-0).

Properti ini bertindak sebagai identitas permanen. Saat produser yang baru dinyalakan mendaftarkan transactional.id yang sama, broker akan mengenali sesi lamanya, membatalkan transaksi menggantung yang belum komit dari sesi lama (epoch fencing), dan mengizinkan produser baru memulai transaksi bersih.

Mekanisme Transaksi: Alur read-process-write #

Berikut adalah 7 langkah kronologis bagaimana transaksi dieksekusi di dalam alur pemrosesan data transaksional:

Inisialisasi (initTransactions): Produser menghubungi Transaction Coordinator untuk mendaftarkan transactional.id. Coordinator akan mengalokasikan Producer ID (PID) baru dan menaikkan nilai epoch untuk memblokir produser lama yang menggantung (zombie fencing).
Mulai Transaksi (beginTransaction): Produser menandai dimulainya transaksi baru secara lokal di memori klien.
Mengirim Pesan (send): Produser mengirimkan pesan ke topik output. Sebelum menulis data ke partisi fisik, produser memberi tahu Coordinator untuk mencatat partisi-partisi tersebut ke dalam jurnal __transaction_state. Langkah ini mencegah broker menulis data dari transaksi yang tidak terdaftar.
Mengirim Offset (sendOffsetsToTransaction): Alih-alih mengirim offset komit langsung ke topik __consumer_offsets secara terpisah, produser mengirimkan offset input tersebut ke Transaction Coordinator. Coordinator akan mendaftarkannya ke dalam transaksi aktif.
Komit Transaksi (commitTransaction): Produser meminta Coordinator untuk menyelesaikan transaksi. Coordinator menulis status PrepareCommit ke jurnal __transaction_state.
Penulisan Penanda Transaksi (Transaction Markers): Coordinator menulis dokumen penanda khusus bernama Commit Marker atau Abort Marker ke seluruh partisi topik output dan topik offset yang berpartisipasi dalam transaksi tersebut.
Selesai: Setelah seluruh penanda transaksi sukses ditulis di semua partisi, Coordinator memperbarui status transaksi di jurnal __transaction_state menjadi CompleteCommit. Transaksi selesai sepenuhnya.

State Diagram: Siklus Hidup Transaksi #

Siklus hidup transaksi Kafka bertransisi melalui serangkaian status di bawah pengawasan Transaction Coordinator seperti yang digambarkan dalam diagram state berikut:

stateDiagram-v2
    [*] --> Empty: initTransactions()
    Empty --> Ongoing: beginTransaction()
    Ongoing --> Ongoing: send() / sendOffsets()
    Ongoing --> PrepareCommit: commitTransaction()
    Ongoing --> PrepareAbort: abortTransaction()
    PrepareCommit --> CompleteCommit: Tulis Commit Marker ke Partisi
    PrepareAbort --> CompleteAbort: Tulis Abort Marker ke Partisi
    CompleteCommit --> Empty: Selesai
    CompleteAbort --> Empty: Selesai

Konfigurasi Tingkat Konsumen: isolation.level #

Jaminan atomisitas transaksi di sisi produser tidak akan berguna jika aplikasi konsumen hilir kita langsung memproses setiap pesan yang masuk tanpa memeriksa status transaksi pesan tersebut. Oleh karena itu, jaminan Exactly-Once mewajibkan kita menyelaraskan konfigurasi konsumen melalui properti isolation.level.

Terdapat dua nilai konfigurasi untuk isolation.level:

1. `read_uncommitted` (Default) #

Konsumen akan membaca seluruh pesan yang tertulis di log partisi broker sesuai urutan offset fisiknya. Ini berarti konsumen akan membaca pesan dari transaksi yang sedang berjalan (ongoing) serta pesan dari transaksi yang telah dibatalkan (aborted). Skenario ini sangat berbahaya jika aplikasi kita tidak toleran terhadap data kotor (dirty reads).

2. `read_committed` #

Konsumen hanya akan membaca:

Pesan non-transaksional (pesan biasa tanpa transaksi).
Pesan transaksional dari transaksi yang telah sukses menerima Commit Marker.
Konsumen secara aktif akan melompati pesan-pesan transaksional yang ditandai dengan Abort Marker.

Bagaimana Konsumen Mengetahui Batas Transaksi? #

Broker Kafka menggunakan penanda internal bernama Last Stable Offset (LSO). LSO adalah offset terkecil dari transaksi yang masih berjalan (ongoing). Konsumen yang dikonfigurasi dengan isolation.level=read_committed tidak akan diizinkan membaca pesan di luar batas LSO, meskipun ada pesan non-transaksional baru yang ditulis setelah transaksi ongoing tersebut. Ini memastikan bahwa konsumen tidak akan mendahului proses transaksi yang belum komit.

Implementasi Java: Pola read-process-write Transaksional #

Berikut adalah kode perbandingan antara alur pemrosesan data non-transaksional yang rentan duplikasi (anti-pattern) dengan alur pemrosesan Exactly-Once Semantics yang aman:

// ANTI-PATTERN: Alur pemrosesan data tanpa jaminan transaksional
// Jika crash terjadi setelah send() namun sebelum commitSync(), pesan duplikat akan dikirim ulang
public class VulnerableStreamProcessor {
    public void process(KafkaConsumer<String, String> consumer, KafkaProducer<String, String> producer) {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 1. Proses data bisnis
                String outputValue = transform(record.value());
                
                // 2. Kirim ke topik output
                producer.send(new ProducerRecord<>("output-topic", record.key(), outputValue));
            }
            // ✗ Komit offset secara terpisah. Jika crash sebelum baris ini eksekusi,
            // data di output-topic sudah terlanjur tertulis, memicu duplikasi data di reload berikutnya.
            consumer.commitSync(); 
        }
    }
    private String transform(String in) { return in.toUpperCase(); }
}

// BENAR: Menggunakan Exactly-Once Semantics (EOS) dengan Transaction API
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class SecureStreamProcessor {
    public void process() {
        // 1. Konfigurasi Produser Transaksional
        Properties prodProps = new Properties();
        prodProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        prodProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        prodProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        // ✓ BENAR: transactional.id wajib diset untuk mengaktifkan Transaction API
        prodProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "processor-tx-id-0");
        prodProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // Wajib aktif

        KafkaProducer<String, String> producer = new KafkaProducer<>(prodProps);

        // 2. Konfigurasi Konsumen Read Committed
        Properties consProps = new Properties();
        consProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consProps.put(ConsumerConfig.GROUP_ID_CONFIG, "processor-group");
        consProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        
        // ✓ BENAR: Konsumen wajib diset hanya membaca transaksi yang sukses komit
        consProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        consProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Matikan auto-commit

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consProps);
        consumer.subscribe(Collections.singletonList("input-topic"));

        // Inisialisasi transaksi di coordinator
        producer.initTransactions();

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) continue;

                // ✓ Mulai transaksi baru secara atomik
                producer.beginTransaction();

                try {
                    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

                    for (ConsumerRecord<String, String> record : records) {
                        // Proses transformasi data
                        String outputValue = record.value().toUpperCase();

                        // Kirim data ke topik output dalam transaksi
                        producer.send(new ProducerRecord<>("output-topic", record.key(), outputValue));

                        // Catat offset pesan input berikutnya yang harus dibaca
                        offsetsToCommit.put(
                            new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1)
                        );
                    }

                    // ✓ Kirim komit offset ke coordinator transaksi untuk disertakan dalam transaksi atomik
                    producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

                    // ✓ Komit seluruh operasi: data terkirim dan offset terkomit secara bersamaan
                    producer.commitTransaction();

                } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
                    // Galat fatal yang tidak bisa dipulihkan, kita harus menutup produser
                    producer.close();
                    break;
                } catch (KafkaException e) {
                    // Galat transient, batalkan transaksi dan coba lagi dari offset sebelumnya
                    System.err.println("Gagal memproses batch, rollback transaksi: " + e.getMessage());
                    producer.abortTransaction();
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
}

Batasan dan Peringatan Penting dalam Menggunakan EOS #

Meskipun Exactly-Once Semantics memberikan jaminan keamanan data yang luar biasa kuat, kita harus memahami beberapa batasan arsitekturalnya agar tidak membuat kesalahan desain sistem di produksi:

1. Hanya Berlaku di Dalam Ekosistem Kafka (Kafka-Only Boundary) #

EOS Kafka hanya menjamin operasi atomik yang terjadi di dalam batas internal kluster Kafka.

Batasan: Jika di dalam utas pemrosesan read-process-write kita melakukan penulisan data ke database SQL eksternal (misal: MySQL) atau memanggil API REST pihak ketiga, transaksi Kafka tidak dapat menjamin atomisitas sistem eksternal tersebut. Jika transaksi Kafka di-abort, data yang terlanjur tertulis di database MySQL tidak akan otomatis di-rollback oleh Kafka. Kita masih membutuhkan implementasi pola Outbox Pattern atau penanganan idempotensi manual di sisi database.

2. Overhead Latensi Transaksi #

Menggunakan transaksi menambah beban overhead jaringan dan I/O broker:

Setiap transaksi membutuhkan penulisan jurnal ke topik __transaction_state.
Setiap partisi tujuan harus menerima Commit/Abort Marker.
Konsumen dengan read_committed mengalami sedikit penundaan membaca karena harus menunggu transaksi selesai (committed) hingga menyentuh batas Last Stable Offset (LSO).
Rekomendasi: Jangan melakukan transaksi dengan ukuran batch yang terlalu kecil (misal per 1 pesan). Lakukan pemrosesan batch yang wajar (misal per 100–1000 pesan) untuk membagi beban overhead transaksi ke banyak pesan secara proporsional.

Ringkasan #

Exactly-Once Semantics: Jaminan pemrosesan data tepat satu kali yang memadukan fitur Idempotent Producer dengan transactional API.

read-process-write: Alur pemrosesan data di mana pesan input dibaca, diproses, dan pesan output ditulis ke topik tujuan bersamaan dengan offset input dalam satu transaksi atomik.

Transaction Coordinator: Broker yang ditunjuk khusus untuk mencatat dan memantau status siklus hidup transaksi ke dalam topik internal __transaction_state.

transactional.id: Identitas unik dan persisten produser yang menjamin perlindungan terhadap kemunculan produser ganda (zombie fencing) setelah terjadi restart.

read_committed: Penyetelan konsumen wajib untuk memastikan konsumen hanya memproses pesan transaksional yang sudah sukses menerima commit marker.

Kafka-Only Boundary: Transaksi Kafka tidak mencakup sistem penyimpanan eksternal (seperti database SQL atau API luar) yang berada di luar lingkup broker Kafka.

← Sebelumnya: Idempotent Producer Berikutnya: Over-Partitioning →