Transaction API #

Ketika kita merancang sistem pemrosesan data real-time, tantangan terbesar muncul saat sebuah data harus didistribusikan secara atomik ke berbagai topik dan partisi yang berbeda. Tanpa jaminan transaksi, kegagalan sistem di tengah proses pengiriman data akan menyisakan data parsial yang terkirim setengah-setengah, menghasilkan inkonsistensi data yang sangat sulit untuk diperbaiki secara manual. Untuk menyelesaikan tantangan koordinasi ini, Apache Kafka menyediakan Transaction API (API Transaksi). API ini memungkinkan aplikasi produser bertindak sebagai agen transaksi terdistribusi yang mampu mengirim sekelompok pesan ke berbagai topik secara atomik (semua pesan tersimpan secara permanen atau tidak ada sama sekali). Melalui artikel ini, kita akan membedah arsitektur internal di balik Transaction API, cara kerja komponen Transaction Coordinator, alur siklus hidup transaksi, perlindungan Zombie Fencing, dan implementasi kode Java yang aman untuk produksi.

Arsitektur Internal: Bagaimana Transaksi Kafka Dikoordinasikan? #

Di balik layar, arsitektur transaksi Kafka didasarkan pada variasi protokol klasik Two-Phase Commit (2PC) terdistribusi yang sangat dioptimalkan untuk performa tinggi. Alih-alih mengandalkan manajer transaksi eksternal yang lambat (seperti XA transactions berbasis protokol X/Open yang membutuhkan koordinator pihak ketiga seperti Atomikos atau Bitronix), Kafka menggunakan komponen internal kluster untuk mengelola status transaksi secara mandiri. Perbedaan mendasar ini krusial: pada transaksi XA, jika koordinator eksternal mati di tengah jalan, seluruh database yang terlibat bisa terkunci dalam status menggantung (blocking locks). Di Kafka, karena koordinatornya adalah broker itu sendiri dan replikasinya dikelola oleh log partisi, pemulihan koordinator berjalan sangat cepat melalui mekanisme failover leader partisi standar tanpa mengunci partisi data.

Arsitektur transaksi ini didukung oleh dua komponen utama:

1. Transaction Coordinator (Koordinator Transaksi) #

Serupa dengan Group Coordinator yang bertugas mengelola keanggotaan kelompok konsumen, Transaction Coordinator adalah modul internal yang berjalan di dalam broker Kafka.

  • Pemilihan Koordinator: Setiap produser transaksional wajib mendeklarasikan string konfigurasi unik bernama transactional.id. Kafka memilih broker mana yang akan menjadi koordinator bagi produser tersebut dengan melakukan hashing terhadap nilai transactional.id tersebut, lalu memetakan hasilnya ke partisi dari topik internal log transaksi.
  • Tugas Koordinator: Mengelola alokasi Producer ID (PID), mencatat pendaftaran partisi-partisi topik yang terlibat dalam transaksi aktif, menulis status transaksi ke log status, dan menyebarkan pesan penanda komit (Commit/Abort Markers) ke broker leader partisi tujuan saat transaksi berakhir.

2. Topik Internal __transaction_state #

Semua transisi status transaksi (seperti Ongoing, PrepareCommit, CompleteCommit) dicatat secara permanen ke topik log internal bernama __transaction_state.

  • Topik ini memiliki banyak partisi (secara default 50 partisi) dan dikonfigurasi dengan kebijakan Log Compaction serta Replication Factor tinggi (default 3 di lingkungan produksi) untuk menjamin ketahanan status transaksi dari kegagalan broker tunggal. Properti penting seperti transaction.state.log.replication.factor=3 dan transaction.state.log.min.isr=2 memastikan bahwa setiap perubahan status transaksi terduplikasi dengan aman sebelum koordinator melanjutkan langkah berikutnya.
  • Penulisan ke topik ini bersifat append-only, memastikan latensi penulisan status transaksi sangat cepat. Proses Log Compaction bertugas memangkas log transaksi lama yang sudah sukses terkomit atau dibatalkan, menyisakan hanya status transaksional aktif atau terkomit paling mutakhir untuk menghemat penyimpanan disk broker.
sequenceDiagram
    autonumber
    participant Prod as Produser Client
    participant TC as Transaction Coordinator
    participant TopicLog as Topik __transaction_state
    participant PartLeader as Broker Partisi Tujuan

    Note over Prod, TC: "Langkah 1: Inisialisasi Transaksi"
    Prod->>TC: InitProducerId (transactional.id="tx_prod_01")
    TC->>TopicLog: Catat PID baru & Epoch ke log
    TC-->>Prod: Return PID & Epoch (Fencing zombie lama)

    Note over Prod, PartLeader: "Langkah 2: Proses Pengiriman Transaksional"
    Prod->>TC: AddPartitionsToTxnRequest (Mendaftarkan partisi tujuan)
    TC->>TopicLog: Catat daftar partisi berstatus ONGOING
    Prod->>PartLeader: ProduceRequest (Kirim Data ke Partisi A & B)
    PartLeader->>PartLeader: Tulis Data ke disk (Belum stabil untuk konsumen)

    Note over Prod, PartLeader: "Langkah 3: Komit Transaksi (2-Phase Commit)"
    Prod->>TC: EndTxnRequest (Commit)
    TC->>TopicLog: Catat status PREPARE_COMMIT
    TC->>PartLeader: WriteTxnMarkerRequest (COMMIT MARKER)
    PartLeader->>PartLeader: Tulis Commit Marker ke Log Partisi
    TC->>TopicLog: Catat status COMPLETE_COMMIT (Transaksi Sukses!)
    TC-->>Prod: Return Commit Sukses

Siklus Hidup Transaksi (Transaction Lifecycle) #

Untuk menggunakan Transaction API secara benar, produser kita harus berinteraksi dengan koordinator melalui serangkaian panggilan API yang memiliki urutan logis yang ketat. Berikut adalah rincian fungsionalitas dari setiap API tersebut:

1. initTransactions() #

Fungsi ini wajib dipanggil tepat satu kali saat aplikasi produser dinyalakan, sebelum ada data yang dikirimkan.

  • Tugas: Menghubungi Transaction Coordinator untuk mendaftarkan transactional.id. Coordinator akan mengalokasikan PID baru atau mencari PID lama yang terikat dengan ID transaksi tersebut, menaikkan nilai Producer Epoch, dan melakukan pembatalan (abort) terhadap semua transaksi menggantung (hanging transactions) dari sesi produser lama yang belum selesai.

2. beginTransaction() #

Dipanggil setiap kali kita ingin memulai batch transaksi baru.

  • Tugas: Menyetel status transaksi internal produser klien menjadi aktif. Klien tidak mengirimkan sinyal apa pun ke broker coordinator pada tahap ini demi menghemat bandwidth jaringan.

3. send() #

Mengirimkan pesan ke satu atau beberapa topik tujuan.

  • Tugas: Sebelum pesan fisik dikirim ke broker leader partisi tujuan, pustaka klien produser secara otomatis mengirimkan request AddPartitionsToTxnRequest ke Transaction Coordinator untuk mendaftarkan partisi tersebut ke daftar transaksi aktif. Setelah terdaftar di __transaction_state, data fisik baru dikirimkan ke broker tujuan.

4. sendOffsetsToTransaction() #

Fungsi khusus yang digunakan jika aplikasi kita bertindak sebagai pemroses stream (Read-Process-Write).

  • Tugas: Mengirimkan offset komit konsumen langsung ke Transaction Coordinator untuk disimpan ke topik internal __consumer_offsets sebagai bagian dari transaksi produser. Ini memastikan bahwa pembacaan pesan lama dan pengiriman pesan baru selesai secara bersamaan dalam satu transaksi atomik tunggal. Di balik layar, coordinator mendaftarkan partisi topik __consumer_offsets tersebut ke dalam daftar transaksi, serupa dengan partisi data biasa.

5. commitTransaction() #

Dipanggil saat seluruh proses pengiriman data sukses tanpa hambatan.

  • Tugas: Mengirim request EndTxnRequest dengan status COMMIT ke koordinator. Koordinator akan menulis status PrepareCommit, menyebarkan Commit Marker ke broker-broker partisi tujuan, dan akhirnya menutup transaksi dengan menulis status CompleteCommit.

6. abortTransaction() #

Dipanggil di dalam blok catch apabila terjadi error atau kegagalan logika bisnis.

  • Tugas: Mengirim request EndTxnRequest dengan status ABORT ke koordinator. Koordinator menulis status PrepareAbort, menyebarkan Abort Marker ke partisi tujuan agar pesan transaksional dibuang secara senyap oleh konsumen read_committed, lalu menulis status CompleteAbort.

Tantangan Batas Waktu Transaksi (Transaction Timeout) #

Setiap transaksi di Kafka dibatasi oleh durasi waktu tertentu untuk mencegah transaksi menggantung selamanya akibat produser yang crash tanpa sempat memanggil abortTransaction(). Batas ini diatur oleh properti transaction.timeout.ms (default 60000 ms atau 1 menit). Jika batas waktu ini terlampaui saat status transaksi masih Ongoing, koordinator secara sepihak akan membatalkan transaksi tersebut di dalam topik __transaction_state dan menyebarkan Abort Marker. Produser yang mencoba melanjutkan pengiriman data pada transaksi yang sama akan menerima TransactionTimeoutException.

Integrasi dengan Kafka Streams #

Bagi pengembang yang menggunakan framework Kafka Streams, Transaction API ini dikelola secara otomatis di bawah kap mesin. Kita cukup menyalakan parameter konfigurasi processing.guarantee="exactly_once_v2". Kafka Streams akan secara dinamis membuat instansi produser transaksional untuk setiap utas pemrosesan (stream thread), mengelola transactional.id unik berbasis ID tugas (task ID), dan secara berkala memanggil siklus komit transaksional beginTransaction() dan commitTransaction() setelah setiap batch pemrosesan data selesai dihitung.

Konsep Zombie Fencing: Pelindung dari Split-Brain #

Dalam arsitektur sistem terdistribusi, salah satu kegagalan yang paling sulit dideteksi adalah masalah Split-Brain atau kemunculan Zombie Producer.

  • Skenario: Bayangkan sebuah produser transaksional Prod_A sedang berjalan. Tiba-tiba, server JVM mengalami jeda Garbage Collection (GC Pause) yang sangat lama (misalnya 1 menit).
  • Dampak: Transaction Coordinator menganggap Prod_A telah mati karena tidak merespons. Sistem orkestrasi (seperti Kubernetes) menyalakan instansi produser baru Prod_B dengan transactional.id yang sama untuk mengambil alih tugas.
  • Masalah: Setelah Prod_B aktif dan memulai transaksi baru, jeda GC pada Prod_A selesai. Prod_A (yang sekarang menjadi zombie) terbangun dan mencoba mengirimkan sisa pesan transaksinya yang tertunda ke broker. Jika diizinkan, ini akan merusak konsistensi data.

Kafka mengatasi bahaya zombie ini menggunakan mekanisme Zombie Fencing:

sequenceDiagram
    autonumber
    participant Prod_A as "Prod_A (Epoch 1)"
    participant Prod_B as "Prod_B (Epoch 2)"
    participant Coord as "Coordinator"
    participant Broker as "Broker"
    
    Note over Prod_A: Mengalami GC Pause
    Coord->>Coord: Menganggap Prod_A mati
    Prod_B->>Coord: InitTransactions
    Coord->>Coord: Epoch dinaikkan menjadi 2
    Note over Prod_A: Bangun dari GC
    Prod_A->>Broker: Mengirim data (Epoch 1)
    Broker-->>Prod_A: MENOLAK DATA! (throw ProducerFencedException)

Ketika Prod_B memanggil initTransactions(), Transaction Coordinator menaikkan nilai Epoch (nomor generasi produser) dari 1 menjadi 2 untuk PID tersebut. Ketika zombie Prod_A mencoba mengirimkan data menggunakan Epoch 1, broker leader partisi akan langsung menolak pengiriman tersebut dan mengembalikan error ProducerFencedException. Prod_A menyadari ia telah digantikan, lalu menghentikan operasinya secara bersih.

Kode Penerapan Java SDK: Transaksi Atomik yang Aman #

Berikut adalah contoh implementasi lengkap penggunaan Transaction API menggunakan Java SDK. Contoh ini mendemonstrasikan bagaimana mengonfigurasi parameter transaksi secara benar, melakukan inisialisasi, dan mengelola kegagalan transaksi menggunakan blok try-catch-finally.

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class SecureTransactionApiExample {
    private static final Logger log = LoggerFactory.getLogger(SecureTransactionApiExample.class);
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TRANSACTIONAL_ID = "finance-payment-tx-01";
    private static final String TOPIC_A = "user-balances";
    private static final String TOPIC_B = "transaction-ledger";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // =====================================================================
        // KONFIGURASI KHUSUS TRANSACTION API
        // =====================================================================
        // ✓ transactional.id wajib disetel dan harus unik per instansi produser
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTIONAL_ID);
        
        // ✓ Secara otomatis mengaktifkan idempotensi produser (enable.idempotence=true)
        // dan memaksa acks=all demi jaminan durabilitas maksimum.
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        
        // Batas waktu tunggu transaksi dibatalkan otomatis oleh broker jika menggantung
        props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000"); // 1 Menit

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

        log.info("Langkah 1: Menginisialisasi transaksi dan membunuh zombie produser lama...");
        // initTransactions() menghubungi koordinator, melakukan alokasi PID,
        // meningkatkan epoch, dan membatalkan sisa transaksi gantung yang lama.
        producer.initTransactions();

        try {
            log.info("Langkah 2: Memulai blok transaksi baru...");
            producer.beginTransaction();

            String userId = "usr_404";
            String debitPayload = "{\"userId\":\"" + userId + "\",\"action\":\"DEBIT\",\"amount\":250000}";
            String ledgerPayload = "{\"txId\":\"tx_999\",\"status\":\"SUCCESS\",\"userId\":\"" + userId + "\"}";

            log.info("Langkah 3: Mengirim pesan transaksional ke Topik A...");
            // Partisi untuk TOPIC_A didaftarkan otomatis ke koordinator sebelum data terkirim
            producer.send(new ProducerRecord<>(TOPIC_A, userId, debitPayload));

            log.info("Langkah 4: Mengirim pesan transaksional ke Topik B...");
            // Partisi untuk TOPIC_B didaftarkan otomatis ke koordinator sebelum data terkirim
            producer.send(new ProducerRecord<>(TOPIC_B, userId, ledgerPayload));

            // Simulasi pengecekan logika bisnis tambahan sebelum komit
            if (System.currentTimeMillis() % 2 == 0) {
                // Untuk demo kegagalan acak
                throw new RuntimeException("Simulasi kegagalan logika bisnis.");
            }

            log.info("Langkah 5: Melakukan komit transaksi secara atomik...");
            // Mengirim request EndTxn ke koordinator untuk menulis Commit Marker
            producer.commitTransaction();
            log.info("✓ Transaksi sukses dideklarasikan secara atomik ke kedua topik.");

        } catch (ProducerFencedException e) {
            // EFEK ZOMBIE FENCING:
            // Eksepsi ini terjadi jika instansi produser lain dengan transactional.id
            // yang sama telah aktif dan mengambil alih kepemilikan transaksi.
            log.error("✗ Gagal: Produser ini terdeteksi sebagai Zombie! Menghentikan aplikasi secara paksa.", e);
            // Jangan memanggil abortTransaction() karena koordinator telah menolak produser ini.
            // Langkah terbaik adalah mematikan aplikasi untuk investigasi split-brain.
            System.exit(1);
        } catch (Exception e) {
            log.error("✗ Terjadi kegagalan selama transaksi. Melakukan rollback / abort...", e);
            try {
                // Mengirim request EndTxn ke koordinator untuk menulis Abort Marker
                producer.abortTransaction();
                log.info("✓ Rollback transaksi berhasil diselesaikan di broker.");
            } catch (Exception abortException) {
                log.error("✗ Gagal melakukan abort transaksi di broker: ", abortException);
            }
        } finally {
            log.info("Menutup koneksi produser secara bersih.");
            producer.close();
        }
    }
}

Ringkasan #

  • Transaction API — Antarmuka pemrograman Kafka yang memungkinkan produser mengirim sekelompok pesan ke berbagai partisi topik secara atomik (All-or-Nothing).
  • Transaction Coordinator — Modul internal broker Kafka yang bertugas mengoordinasikan status transaksi klien berdasarkan fungsi hash dari konfigurasi transactional.id.
  • Topik __transaction_state — Topik log internal terkompresi tempat koordinator mencatat setiap transisi status transaksi sebelum memancarkan sinyal marker.
  • Zombie Fencing — Mekanisme perlindungan untuk memblokir sisa pengiriman dari produser lama (zombie) dengan menolak Epoch yang lebih rendah dari Epoch terdaftar terbaru di broker.
  • Two-Phase Commit (2PC) — Protokol koordinasi internal Kafka yang membagi komit transaksi menjadi fase penulisan catatan PrepareCommit disusul penyebaran Commit Markers.
  • Commit/Abort Markers — Pesan kontrol khusus yang ditulis koordinator ke log partisi fisik sebagai penanda batas akhir transaksi bagi konsumen hilir.
  • initTransactions() — Fungsi inisialisasi wajib untuk mendaftarkan ID transaksi, menaikkan Epoch produser, dan membersihkan transaksi menggantung dari sesi sebelumnya.

← Sebelumnya: Exactly-Once   Berikutnya: Producer Consumer Transaction →

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