Belajar Apache Kafka - Kafka Transactions

#Pendahuluan

Pada episode 8 kalian sudah memahami delivery semantics dan posisi exactly-once dalam spektrum at-most-once dan at-least-once. Sekarang waktunya mempraktikkan bagian yang paling teknis: Kafka transactions — mekanisme yang membuat pengiriman record ke beberapa partition menjadi atomic.

Transaksi Kafka menyelesaikan masalah yang tidak bisa ditangani idempotent producer sendirian. Idempotent producer mencegah duplikasi dalam satu partition, tetapi tidak membuat beberapa tulis ke partition atau topic berbeda menjadi all-or-nothing. Transaksi menutup celah itu dengan memperkenalkan transactional id dan transaction coordinator.

Episode 9 ini akan membawa kalian melalui lima method inti transaction API, pola read-process-write yang menjadi tulang punggung stream processing, konfigurasi yang mengatur perilaku transaksi, serta biaya performa yang harus kalian pertimbangkan sebelum memutuskan memakai transaksi.

#Transaction API

#Lifecycle Lima Method

Transaksi Kafka dijalankan melalui lima method pada KafkaProducer yang mengikuti urutan ketat:

producer.initTransactions();
producer.beginTransaction();
producer.send(new ProducerRecord<>("orders", "order-001"));
producer.send(new ProducerRecord<>("order-events", "order-001"));
producer.commitTransaction();

• initTransactions(): menginisialisasi producer transaksional dan meregistrasi transactional.id ke coordinator.
• beginTransaction(): menandai awal transaksi; semua send setelah ini menjadi bagian transaksi.
• send(): mengirim record seperti biasa, tetapi belum terlihat oleh consumer read_committed sampai commit.
• commitTransaction(): men-flush semua record dan mengakhiri transaksi secara sukses.
• abortTransaction(): membatalkan transaksi; semua record yang dikirim dibuang dan tidak pernah terlihat.

Setiap transaksi harus diakhiri dengan commit atau abort. Membiarkan transaksi menggantung akan membuat coordinator menunda (timeout) dan membatalkannya secara otomatis.

#Menangani Kegagalan

Jangan pernah membiarkan exception membawa transaksi dalam status tidak jelas:

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "order-001", "created"));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction();
}

Jika transaksi difence (producer lain memakai transactional.id yang sama), instance lama harus ditutup. Untuk error lain, abortTransaction() membersihkan status agar coordinator tidak menahan resource.

#Transactional Patterns

#Read-Process-Write

Pola paling umum: consumer membaca record, aplikasi memprosesnya, lalu producer menulis hasilnya — semuanya dalam satu transaksi. Hasil proses dan offset konsumsi dikirim bersama, sehingga rollback salah satunya membatalkan keduanya:

consumer.subscribe(List.of("input-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    producer.beginTransaction();
    for (ConsumerRecord<String, String> record : records) {
        producer.send(new ProducerRecord<>("output-topic", record.key(), "processed:" + record.value()));
        producer.sendOffsetsToTransaction(currentOffsets(records), consumer.groupMetadata());
    }
    producer.commitTransaction();
}

Perhatikan sendOffsetsToTransaction: offset dikirim sebagai bagian transaksi, sehingga konsumsi ulang tidak terjadi ketika proses gagal — inilah dasar end-to-end exactly-once di Kafka Streams.

#Atomic Multi-Partition Writes

Satu transaksi bisa menulis ke partition dan topic mana pun. Misalnya, service order menulis ke orders dan inventory-updates sekaligus; keduanya terlihat oleh consumer read_committed hanya ketika keduanya berhasil di-commit. Kegagalan di salah satu akan meng-abort semuanya.

#Exactly-Once Stream Processing

Pada stream processing, transaksi menggabungkan tulis state internal (changelog) dengan tulis output. Kafka Streams memakai pola ini secara otomatis dengan processing.guarantee=exactly_once_v2 — kalian tidak perlu menulis kode transaksi manual, library yang menanganinya.

#Transaction Configuration

#transactional.id

transactional.id harus unik dan stabil per instance aplikasi. Nilai yang stabil memungkinkan coordinator mendeteksi producer lama yang sudah mati dan menolak tulisannya (fencing):

transactional.id=order-processor-1
transaction.timeout.ms=60000
enable.idempotence=true
acks=all

Jika dua instance memakai transactional.id yang sama secara bersamaan, instance kedua akan mencekal (fencing) yang pertama — mencegah dua proses menulis secara simpang siur. Karena itu pastikan transactional.id unik per instance, misalnya order-processor-1 untuk node pertama dan order-processor-2 untuk node kedua.

#isolation.level di Consumer

Consumer memilih apakah transaksi yang belum selesai boleh dibaca:

• read_uncommitted (default): melihat semua record, termasuk yang masih dalam transaksi atau sudah di-abort.
• read_committed: hanya melihat record dari transaksi yang sudah commit; record dari transaksi di-abort dibuang.

Konsistensi end-to-end hanya tercapai bila producer memakai transaksi dan consumer memakai isolation.level=read_committed secara bersamaan.

#Performance Considerations

#Biaya Transaksi

Transaksi bukan sesuatu yang gratis. Setiap transaksi melibatkan setidaknya satu round-trip tambahan ke transaction coordinator, dan commit membutuhkan penulisan marker di topic internal __transaction_state. Dampaknya:

• Throughput: turun dibanding producer non-transaksional, terutama untuk transaksi kecil dan sering.
• Latensi: naik karena menunggu sinkronisasi status transaksi antar partition.
• Resource: coordinator menahan memori dan disk untuk status transaksi yang sedang berjalan.

#Trade-Off yang Bijak

Jangan membungkus setiap record dalam transaksi sendiri. Praktik terbaik:

• Batch banyak record dalam satu transaksi — misalnya satu transaksi per batch konsumsi, bukan per record.
• Pakai transaksi hanya untuk alur yang benar-benar membutuhkan atomicity: read-process-write, sinkronisasi state kritis, atau sinkronisasi antar topic.
• Untuk pipeline sederhana yang toleran duplikat, at-least-once tetap jauh lebih murah.
• Pantau transaction.count dan metrik coordinator untuk mendeteksi transaksi yang menggantung.

#Penutup

Di episode 9 ini kalian sudah menjalankan lima method transaction API, memahami pola read-process-write dengan sendOffsetsToTransaction, mengonfigurasi transactional.id dan isolation.level, serta menimbang biaya throughput dan latensi dari transaksi.

Inti yang harus dibawa pulang:

• Transaksi membuat tulis lintas partition atomic: semua terlihat atau tidak sama sekali.
• Lifecycle-nya selalu initTransactions, beginTransaction, send, lalu commit atau abort.
• sendOffsetsToTransaction mengikat offset konsumsi ke transaksi untuk EOS end-to-end.
• transactional.id harus unik dan stabil; fencing mencegah dua instance menulis bersamaan.
• Consumer wajib isolation.level=read_committed untuk konsistensi penuh.
• Batch banyak record per transaksi; jangan transaksi per record.

Di episode 10 selanjutnya kita akan membahas log compaction — kebalikan dari retention delete. Kalian akan belajar cara Kafka mempertahankan nilai terbaru per key dengan tombstone, parameter min.compaction.lag.ms, max.compaction.lag.ms, dan min.cleanable.dirty.ratio, serta use case changelog topic, materialized views, dan state store. Transaksi dan compaction adalah dua fondasi penyimpanan state Kafka yang akan terus kalian temui.