Producer Consumer Transaction #

Dalam arsitektur mikro (microservices) berbasis peristiwa (event-driven), pola pemrosesan yang paling sering kita temui adalah pola Read-Process-Write (Baca-Proses-Tulis). Dalam pola ini, sebuah aplikasi mengonsumsi pesan dari sebuah topik masukan (input topic), melakukan beberapa transformasi logika bisnis atau penghitungan data, lalu mengirimkan hasilnya ke topik keluaran (output topic) lain. Untuk menjamin keandalan sistem ini secara menyeluruh, kita harus memastikan bahwa pembacaan pesan asal, penulisan pesan baru, dan pembaruan posisi baca (komit offset) berhasil dieksekusi secara atomik sebagai satu kesatuan unit transaksi. Di sinilah integrasi transaksi antara produser dan konsumen di Apache Kafka memainkan peran vital. Artikel ini akan membedah secara mendalam bagaimana merancang transaksi Read-Process-Write yang aman, mengapa metode komit offset biasa tidak berlaku di sini, bagaimana mengalirkan offset ke dalam transaksi produser, serta contoh implementasi kode Java yang siap pakai di lingkungan produksi.

Tantangan Pola Read-Process-Write dan Mengapa Komit Biasa Gagal #

Pada aplikasi non-transaksional, siklus pemrosesan Read-Process-Write beroperasi dengan memisahkan transaksi di sisi konsumen dan produser:

Konsumen membaca pesan dengan offset 300 dari topik input-orders.
Aplikasi memproses pesan tersebut.
Produser mengirimkan pesan baru hasil pemrosesan ke topik output-invoices.
Konsumen mengirimkan komit offset 300 ke broker secara terpisah.

Di mana titik kegagalannya? Jika langkah ke-3 sukses (pesan invoice terkirim ke broker), namun server mengalami crash di langkah ke-4 sebelum offset 300 dikomit. Saat aplikasi pulih, konsumen akan membaca kembali pesan offset 300 dari topik input-orders, memprosesnya lagi, dan produser akan mengirimkan invoice duplikat ke topik output-invoices. Ini merusak semantik exactly-once.

Untuk mengatasi kegagalan parsial ini, kita tidak boleh menggunakan metode komit offset standar (seperti consumer.commitSync() atau auto-commit). Sebaliknya, offset konsumen harus dikomit di dalam transaksi produser. Melalui cara ini, broker koordinator akan memperlakukan penulisan offset ke topik internal __consumer_offsets sama seperti penulisan pesan biasa ke topik keluaran. Jika transaksi sukses, pesan baru dan offset baru akan terkomit secara bersamaan. Jika transaksi dibatalkan (aborted), pesan baru tidak akan terlihat oleh konsumen hilir dan offset konsumen masukan tidak akan bergeser, membiarkan aplikasi kita memproses ulang pesan masukan tersebut secara aman dari posisi semula.

flowchart TD
    subgraph Input ["1. Tahap Pembacaan (Read)"]
        Cons["Kafka Consumer"] -->|"Tarik data (Offset 100)"| InTopic["Input Topic: orders"]
    end
    
    subgraph Process ["2. Tahap Pemrosesan"]
        App["Utas Aplikasi (Transformasi Data)"]
    end
    
    subgraph Output ["3. Tahap Penulisan Atomik (Write)"]
        Prod["Kafka Producer (Transactional)"]
        OutTopic["Output Topic: invoices"]
        OffsetTopic["Internal Topic: __consumer_offsets"]
        
        Prod -->|"A. Kirim Invoice Baru"| OutTopic
        Prod -->|"B. Kirim Commit Offset 100"| OffsetTopic
    end
    
    InTopic --> Cons
    Cons --> App
    App --> Prod
    
    style InTopic stroke:#333,stroke-width:2px
    style OutTopic stroke:#2e7d32,stroke-width:2px
    style OffsetTopic stroke:#0288d1,stroke-width:2px

Mengalirkan Offset ke dalam Transaksi Produser #

Kunci utama dari atomisitas transaksi Read-Process-Write terletak pada fungsi API produser:

producer.sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata)

Fungsi ini memberi tahu Transaction Coordinator untuk mendaftarkan offset konsumen ke dalam transaksi aktif yang sedang berjalan.

Mengapa Memerlukan ConsumerGroupMetadata? #

Sejak Apache Kafka 2.5, parameter ConsumerGroupMetadata (yang diperoleh melalui consumer.groupMetadata()) wajib disertakan dalam panggilan API ini.

Tujuan: Parameter ini membawa informasi penting seperti Member ID konsumen dan nilai generasi kelompok konsumen (Generation ID).
Zombie Fencing di Sisi Konsumen: Informasi metadata ini digunakan oleh broker koordinator untuk melakukan verifikasi bahwa instansi konsumen yang meminta komit offset adalah anggota grup yang sah. Jika terjadi rebalance dan konsumen tersebut telah dikeluarkan dari grup (karena mengalami jeda GC lama) dan posisinya digantikan oleh konsumen baru, koordinator akan mendeteksi perbedaan Generation ID dan menolak komit offset tersebut dengan melempar FencedInstanceIdException atau CommitFailedException. Ini mencegah zombie konsumen mengacaukan koordinasi grup.

Detail Mekanisme Protokol Fencing #

Saat sendOffsetsToTransaction dieksekusi, pustaka klien produser mengirimkan request TxnOffsetCommitRequest ke broker Group Coordinator (broker yang mengelola kelompok konsumen kita, bukan Transaction Coordinator). Request ini membungkus offset data, ID grup, Member ID, dan Generation ID. Group Coordinator memvalidasi parameter tersebut:

Jika Generation ID yang dikirim oleh produser atas nama konsumen lebih kecil dari Generation ID aktif kelompok konsumen saat ini di broker, koordinator menyadari bahwa rebalance telah selesai dan konsumen ini sudah kedaluwarsa.
Koordinator langsung menolak komit offset tersebut dan mengembalikan error FencedState.
Klien produser menangkap error ini dan melemparkan eksepsi transaksional untuk membatalkan seluruh transaksi, memastikan tidak ada data keluaran yang terlanjur terkomit padahal offset masukan gagal dikomit.

Skenario Scaling pada Kafka Streams #

Bagi pengembang yang menggunakan framework Kafka Streams, koordinasi Read-Process-Write terdistribusi ini disederhanakan secara dramatis. Kafka Streams menggunakan parameter processing.guarantee="exactly_once_v2". Di balik layar, Kafka Streams membagi topologi pemrosesan menjadi unit-unit kecil bernama Tasks. Setiap Task ditugaskan untuk memproses partisi input tertentu. Untuk menjamin EOS, Kafka Streams mengalokasikan produser transaksional terpisah untuk setiap Stream Thread, dengan transactional.id yang diformat secara terstruktur: <application.id>-<task.id>. Format statis yang unik per Task ini krusial: jika satu task mengalami crash dan dijadwalkan ulang pada thread atau server lain, instansi task baru akan menginisialisasi produser dengan transactional.id yang sama. Inisialisasi ini secara otomatis membunuh (fences) instansi task lama yang mungkin masih berjalan (zombie) berkat mekanisme epoch fencing di Transaction Coordinator.

Perilaku Isolation Level read_committed di Konsumen Hilir #

Agar data yang diproses secara transaksional dapat dibaca dengan aman oleh sistem hilir, aplikasi konsumen hilir (downstream consumers) yang membaca dari output-invoices wajib disetel menggunakan properti isolation.level=read_committed.

Jika konsumen hilir menggunakan default (read_uncommitted), mereka akan langsung membaca pesan baru di topik keluaran segera setelah pesan tersebut ditulis oleh produser, tanpa menunggu transaksi selesai dikomit. Hal ini melanggar jaminan atomisitas karena jika transaksi produser di atas akhirnya dibatalkan (aborted), konsumen hilir terlanjur memproses data kotor yang salah (dirty reads).

Visualisasi Perbandingan Pembacaan Offset #

Mari kita lihat struktur log partisi berikut untuk memahami batas-batas pembacaan antara kedua isolation level:

STRUKTUR LOG PARTISI:
Offset:  │ 100   │ 101   │ 102   │ 103   │ 104   │ 105   │ 106
Pesan:   │ Msg A │ Msg B │ Msg C │ Msg D │ Msg E │ Msg F │ COMMIT marker
Status:  │ Normal│ Tx_01 │ Tx_02 │ Tx_01 │ Tx_02 │ Normal│ untuk Tx_01
         └───────┴───────┴───────┴───────┴───────┴───────┴──────────────
                 ▲                                       ▲
                 │                                       │
                LSO                                     HW (High Watermark)

Dalam skenario di atas:

Tx_01 adalah transaksi yang sukses dikomit di offset 106.
Tx_02 (Msg C di offset 102 dan Msg E di offset 104) adalah transaksi yang masih berjalan (ongoing) dan belum selesai.
High Watermark (HW) adalah offset terakhir yang berhasil ditulis ke semua replika (offset 106).
Last Stable Offset (LSO) berada di offset 101 karena Msg B (Tx_01) dan Msg C (Tx_02) bertumpuk, dan Tx_02 masih aktif berjalan. LSO tidak dapat bergerak melewati offset 102.

Bagaimana konsumen membaca data ini?

Konsumen read_uncommitted: Akan membaca semua pesan dari offset 100 hingga 106 (termasuk data transaksi ongoing Tx_02).
Konsumen read_committed: Hanya akan membaca Msg A di offset 100. Konsumen tertahan tidak bisa membaca Msg B (offset 101) atau Msg F (offset 105) karena log diblokir oleh LSO di offset 102 yang menunggu kejelasan status Tx_02. Setelah Tx_02 menerima commit/abort marker, LSO akan meloncat maju dan konsumen langsung dapat melanjutkan pembacaan.

Konsep Last Stable Offset (LSO) dan Hambatan LSO #

Di sisi broker, untuk setiap partisi topik, Kafka memelihara batas pointer pembacaan bernama Last Stable Offset (LSO).

Definisi: LSO adalah offset terkecil dari transaksi yang saat ini masih berstatus berjalan (ongoing/active).
Perilaku Konsumen: Konsumen dengan read_committed hanya diizinkan membaca pesan hingga batas LSO. Jika ada transaksi yang sedang berjalan lama dan belum dikomit pada suatu partisi, LSO partisi tersebut akan tertahan di posisi awal transaksi tersebut dimulai. Meskipun ada ratusan pesan non-transaksional baru yang ditulis setelah itu, konsumen read_committed akan tertahan (block/pause) di batas LSO dan tidak dapat membaca pesan baru tersebut sampai transaksi ongoing tadi selesai dikomit atau dibatalkan. Fenomena ini disebut LSO Blockage (Hambatan LSO). Oleh karena itu, kita harus menjaga durasi setiap transaksi sesingkat mungkin.

Kode Penerapan Java: Implementasi Read-Process-Write #

Berikut adalah contoh implementasi lengkap pola pemrosesan transaksi Read-Process-Write menggunakan Java SDK. Kode ini secara aman menangani siklus hidup transaksi, mengalirkan offset ke dalam transaksi produser, menyertakan metadata grup konsumen, dan melakukan rollback jika terjadi kegagalan pemrosesan.

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

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TransactionalStreamProcessor {
    private static final Logger log = LoggerFactory.getLogger(TransactionalStreamProcessor.class);
    private static final String INPUT_TOPIC = "input-orders";
    private static final String OUTPUT_TOPIC = "output-invoices";
    private static final String TRANSACTIONAL_ID = "stream-processor-tx-01";
    private static final String GROUP_ID = "orders-processor-group";

    public static void main(String[] args) {
        // 1. Konfigurasi Konsumen Masukan
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        
        // Wajib matikan auto-commit karena kita akan mengalirkan offset ke dalam transaksi produser
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Gunakan read_committed jika konsumen ini juga membaca dari topik transaksional lain
        consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

        // 2. Konfigurasi Produser Transaksional
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // transactional.id wajib unik per instansi produser aktif
        producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTIONAL_ID);
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

        consumer.subscribe(Collections.singletonList(INPUT_TOPIC));

        log.info("Menginisialisasi transaksi produser...");
        producer.initTransactions();

        try {
            while (true) {
                // Tarik data dari topik input
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                
                if (!records.isEmpty()) {
                    log.info("Menerima {} records untuk diproses.", records.count());
                    
                    try {
                        // ✓ LANGKAH A: Memulai transaksi baru di sisi produser
                        producer.beginTransaction();

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

                        for (ConsumerRecord<String, String> record : records) {
                            // ✓ LANGKAH B: Proses data (Logika Bisnis)
                            String outputValue = transformOrderToInvoice(record.value());
                            
                            // ✓ LANGKAH C: Kirim hasil pemrosesan ke topik output
                            ProducerRecord<String, String> outRecord = new ProducerRecord<>(
                                    OUTPUT_TOPIC, record.key(), outputValue
                            );
                            producer.send(outRecord);

                            // Catat posisi offset berikutnya (offset saat ini + 1) untuk setiap partisi
                            offsetsToCommit.put(
                                    new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1)
                            );
                        }

                        // ✓ LANGKAH D: Alirkan offset komit konsumen ke dalam transaksi produser.
                        // Sertakan metadata grup konsumen untuk mengamankan proses dari zombie konsumen.
                        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

                        // ✓ LANGKAH E: Komit transaksi secara atomik (Pesan baru + Offset komit disimpan bersamaan)
                        producer.commitTransaction();
                        log.info("✓ Transaksi berhasil dikomit secara atomik.");

                    } catch (ProducerFencedException e) {
                        // Terjadi split-brain: Produser ini dideteksi sebagai zombie
                        log.error("✗ Fatal: Produser terdeteksi sebagai zombie. Menghentikan proses.", e);
                        break;
                    } catch (Exception e) {
                        // Terjadi error logika bisnis atau kesalahan jaringan biasa
                        log.error("✗ Gagal memproses batch. Melakukan rollback transaksi...", e);
                        // Membatalkan penulisan pesan dan membatalkan pembaruan offset konsumen
                        producer.abortTransaction();
                    }
                }
            }
        } finally {
            log.info("Menutup koneksi konsumen dan produser secara bersih.");
            consumer.close();
            producer.close();
        }
    }

    private static String transformOrderToInvoice(String orderJson) throws Exception {
        // Simulasi logika bisnis transformasi data
        if (orderJson.contains("ERROR_TRIGGER")) {
            throw new Exception("Simulasi error pemrosesan data.");
        }
        return orderJson.replace("ORDER", "INVOICE");
    }
}

Kapan Menggunakan Pola Transaksi Read-Process-Write? #

Gunakan panduan keputusan berikut saat menentukan arsitektur transaksi sistem Anda:

TETAP gunakan Read-Process-Write Transaksional jika:
  ✓ Aliran pipa pemrosesan data (pipeline) sepenuhnya berada di dalam Kafka (Kafka-to-Kafka).
  ✓ Data dihitung secara berantai melintasi beberapa topik dan partisi (misalnya agregasi e-commerce).
  ✓ Anda menggunakan Kafka Streams API dan ingin menjamin pemrosesan stateful (seperti windowing join) bebas duplikasi.

JANGAN gunakan pola ini (Gunakan manual deduplikasi database) jika:
  ✗ Logika pemrosesan data melibatkan penulisan ke database eksternal (SQL/NoSQL) secara langsung.
  ✗ Aplikasi melakukan panggilan HTTP REST API ke pihak ketiga (seperti payment gateway) di tengah proses.
  ✗ Anda ingin menghindari latensi overhead koordinasi Two-Phase Commit kluster Kafka.

Ringkasan #

Read-Process-Write (RPW) — Pola pemrosesan stream terdistribusi di mana membaca dari topik input, memproses, dan mengirim hasil ke topik output dikelola dalam satu transaksi tunggal.

Offset dalam Transaksi — Komit offset konsumen masukan harus dikirim melalui fungsi produser sendOffsetsToTransaction() agar terkomit secara atomik bersama dengan pesan keluaran baru.

ConsumerGroupMetadata — Parameter wajib dalam komit transaksi untuk mengamankan sistem dari zombie konsumen melalui pencocokan Generation ID aktif di Group Coordinator.

LSO (Last Stable Offset) — Batas offset terendah dari transaksi terlama yang masih aktif pada suatu partisi topik, menjadi pembatas pembacaan bagi konsumen hilir.

LSO Blockage (Hambatan LSO) — Dampak negatif di mana konsumen read_committed tertahan tidak dapat membaca pesan baru akibat adanya transaksi transaksional lain yang berjalan terlalu lama.

Zero Data Loss & Zero Duplication — Keunggulan utama pola RPW transaksional yang memastikan tidak ada pesan masukan yang terlewatkan dan tidak ada pesan keluaran duplikat yang dipancarkan ke topik hilir.

← Sebelumnya: Exactly-Once Berikutnya: EOS Limitation →