Transaksi

Transaksi producer menjadikan sekumpulan pesan ke beberapa partisi satu kesatuan tulisan, yaitu semuanya menjadi terbaca bagi konsumen atau semuanya tersembunyi. Topik ini membahas siklus hidup transaksi dari init_transactions sampai commit_transaction, tingkat isolasi read_committed dan read_uncommitted di sisi konsumen, serta pola consume-transform-produce yang menjadikan hasil transformasi dan commit offset konsumen satu kesatuan. Demo dijalankan dengan kafka-python pada broker lokal dan memperlihatkan perbedaan kedua tingkat isolasi pada topic yang sama.

Siklus hidup transaksi producer#

Producer masuk mode transaksional begitu dikonfigurasi dengan transactional.id, yaitu pengenal yang tetap digunakan oleh satu instance producer antar sesi. Konfigurasi tersebut otomatis mengaktifkan pengiriman idempoten, karena transaksi dibangun di atas nomor urut idempoten yang dibahas pada topik Ketahanan producer. Producer tanpa transactional.id tetap dapat menjadi idempoten, tetapi seluruh panggilan transaksi menuntut pengenal ini dan menolak berjalan tanpanya.

Tabel berikut memetakan panggilan transaksi pada klien Java ke padanannya pada kafka-python yang digunakan dalam demo, beserta peran masing-masing.

Urutan penggunaannya berpatokan pada lima panggilan itu. init_transactions dipanggil sekali sebelum transaksi pertama, kemudian setiap transaksi diawali begin_transaction, diisi pengiriman pesan, dan ditutup dengan commit_transaction atau abort_transaction. Kesalahan di tengah transaksi ditangani dengan menutup transaksinya secara eksplisit, bukan membiarkannya menggantung, karena partisi yang menyimpan transaksi terbuka menahan pembacaan read_committed pada posisi yang disebut last stable offset (LSO), yaitu offset sebelum transaksi pertama yang belum selesai.

Kepemilikan transactional.id dijaga oleh broker dengan mekanisme fencing. Producer baru yang memanggil init_transactions dengan pengenal yang sama memperoleh epoch lebih tinggi, dan producer lama yang mencoba mengirim atau commit setelahnya ditolak dengan error ProducerFencedException pada klien Java atau ProducerFencedError pada kafka-python. Berkat aturan itu, producer yang berhenti karena crash tidak menyisakan tulisan setelah penggantinya bangun, dan transaksi yang menggantung dari sesi lama di-abort otomatis saat init_transactions dijalankan.

Isolasi read_committed dan read_uncommitted#

Isolasi adalah sisi konsumen dari transaksi, diatur lewat konfigurasi isolation.level pada consumer. Nilai default read_uncommitted mengembalikan semua pesan tanpa memandang transaksinya, termasuk pesan dari transaksi yang sudah di-abort. Nilai read_committed menyembunyikan pesan yang dibatalkan, sehingga konsumen hanya melihat hasil transaksi yang commit. Pesan nontransaksional dikembalikan di kedua mode tanpa penyaringan apa pun.

Penyaringan ditegakkan oleh marker, yaitu pesan kontrol yang ditulis broker ke setiap partisi yang disentuh transaksi pada saat commit atau abort. Marker menempati offset di log seperti pesan biasa, tetapi tidak pernah dikembalikan ke konsumen mana pun. Karena itu sebuah partisi dapat menampilkan celah offset pada pembacaan, dan celah itu ditempati marker, bukan pesan yang tersembunyi saja.

Pembacaan read_committed berhenti sampai LSO, sehingga transaksi yang lambat atau menggantung menahan pesan setelahnya meskipun pesan itu tidak bagian dari transaksi apa pun. Aturan ini menjaga kepastian hasil transaksi yang belum diputuskan, karena konsumen tidak boleh melihat pesan yang mungkin di-abort kemudian. Bagi pemantauan, transaksi terbuka yang berlarut tampil sebagai lag yang membeku pada konsumen read_committed, sehingga durasi transaksi layak dijaga tetap pendek.

Demo transaksi dan dua tingkat isolasi#

Skrip berikut menulis ke topic unik dengan dua partisi dalam dua transaksi. Transaksi pertama menulis satu bacaan lalu dibatalkan, dan transaksi kedua menulis tiga bacaan ke dua partisi lalu commit. Setelah producer selesai, dua konsumen membaca topic yang sama, yang pertama dengan read_uncommitted dan yang kedua dengan read_committed, masing-masing tanpa group agar kedua partisi dibaca dari awal.

from kafka import KafkaConsumer, KafkaProducer, KafkaAdminClient, TopicPartition
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError
import time

topic = f"m4-transaksi-{int(time.time())}"  # nama unik per eksekusi

admin = KafkaAdminClient(bootstrap_servers="localhost:9092")
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=2, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()

producer = KafkaProducer(bootstrap_servers="localhost:9092",
                         transactional_id="m4-producer-transaksi")
producer.init_transactions()

# transaksi pertama: satu bacaan yang pada akhirnya dibatalkan
producer.begin_transaction()
producer.send(topic, value=b"bacaan-janggal", partition=0)
producer.flush()
producer.abort_transaction()
print("transaksi 1: bacaan-janggal ditulis lalu dibatalkan")

# transaksi kedua: tiga bacaan ke dua partisi dalam satu transaksi
producer.begin_transaction()
producer.send(topic, value=b"bacaan-0", partition=0)
producer.send(topic, value=b"bacaan-1", partition=1)
producer.send(topic, value=b"bacaan-2", partition=0)
producer.flush()
producer.commit_transaction()
print("transaksi 2: tiga bacaan commit dalam satu transaksi")
producer.close()

# dua konsumen membaca topic yang sama dengan tingkat isolasi berbeda
p0, p1 = TopicPartition(topic, 0), TopicPartition(topic, 1)
for level in ("read_uncommitted", "read_committed"):
    consumer = KafkaConsumer(bootstrap_servers="localhost:9092",
                             auto_offset_reset="earliest",
                             isolation_level=level,
                             consumer_timeout_ms=2000)
    consumer.assign([p0, p1])
    for msg in consumer:
        print(f"{level} partisi={msg.partition} offset={msg.offset} {msg.value.decode()}")
    consumer.close()

transaksi 1: bacaan-janggal ditulis lalu dibatalkan
transaksi 2: tiga bacaan commit dalam satu transaksi
read_uncommitted partisi=0 offset=0 bacaan-janggal
read_uncommitted partisi=0 offset=2 bacaan-0
read_uncommitted partisi=0 offset=3 bacaan-2
read_uncommitted partisi=1 offset=0 bacaan-1
read_committed partisi=0 offset=2 bacaan-0
read_committed partisi=0 offset=3 bacaan-2
read_committed partisi=1 offset=0 bacaan-1

Keluaran read_uncommitted memuat bacaan-janggal pada offset 0, sedangkan read_committed tidak memuatnya sama sekali. Celah pada offset 1 partisi 0 ditempati marker abort, dan offset 1 pada partisi 1 ditempati marker commit. Keduanya tidak pernah tampil di pembacaan mana pun. Tiga bacaan transaksi kedua menempati offset berurutan pada dua partisi dan semuanya terbaca setelah marker commit ditulis, sekaligus memperlihatkan penulisan atomik ke dua partisi dalam satu transaksi.

Pola consume-transform-produce#

Pola consume-transform-produce membaca dari topic masukan, mengubah isinya, dan menulis hasilnya ke topic tujuan. Tanpa transaksi, hasil transformasi dan commit offset masukan adalah dua tulisan yang berdiri sendiri. Aplikasi yang berhenti di antara keduanya meninggalkan pesan yang sudah diproses tetapi belum di-commit, sehingga pesan yang sama diproses ulang, atau sebaliknya commit yang terlanjur maju tanpa hasil yang tersimpan. Transaksi menutup celah di antara keduanya dengan memasukkan commit offset ke dalam transaksi lewat send_offsets_to_transaction.

Syarat pola ini berkaitan dengan sisi konsumennya. Commit otomatis dimatikan dengan enable.auto.commit=false karena posisi offset kini ditentukan oleh transaksi, bukan timer konsumen. Offset yang dikirim adalah posisi konsumen setelah pesan masukan diproses, dan pesan tujuan ditulis sebelum commit_transaction dipanggil sehingga keduanya commit sebagai satu kesatuan. Konsumen pada topic tujuan membaca dengan read_committed agar hasil dari transaksi yang di-abort tidak pernah terbaca, misalnya hasil transaksi yang diulang setelah aplikasi mati.

Kerangka panggilannya berbentuk siklus seperti berikut. Potongan ini adalah fragmen yang masih memerlukan variabel topic, fungsi transformasi, dan penentuan partisinya agar dapat dijalankan.

# tidak-dijalankan
producer.init_transactions()
while True:
    batch = konsumen.poll(timeout_ms=500)  # pesan dari topic masukan
    if not batch:
        continue
    producer.begin_transaction()
    for kelompok in batch.values():
        for pesan in kelompok:
            producer.send(topic_tujuan, value=ubah(pesan))  # hasil transformasi
    offset = OffsetAndMetadata(konsumen.position(tp), "", -1)
    producer.send_offsets_to_transaction({tp: offset}, konsumen.group_metadata())
    producer.commit_transaction()  # pesan tujuan dan offset commit bersama-sama

Dokumentasi resmi menambahkan dua kebiasaan ketika pola ini dijalankan pada consumer group yang mengalami rebalancing. Konsumen menjamin keunikannya melalui assignment partisi yang dibahas pada topik Consumer group, dan satu instance producer digunakan untuk satu instance konsumen agar pengenal transaksinya tidak diperebutkan. Dengan read_committed di kedua sisi, pemrosesan ulang yang sah tetap berlangsung tanpa duplikat yang terbaca, sehingga pola ini memberikan jaminan tepat sekali sepanjang log Kafka.

Kesalahan umum#

1. Mengirim pesan setelah transaksi ditutup tanpa memulai transaksi baru. Producer transaksional menolak pengiriman di luar transaksi dengan error illegal state, sehingga aplikasi berhenti di tengah alur. Awali setiap gelombang pengiriman dengan begin_transaction dan tutup dengan commit atau abort.
2. Membiarkan commit otomatis aktif pada konsumen dalam pola consume-transform-produce. Offset yang di-commit oleh timer berada di luar transaksi, sehingga aplikasi yang berhenti di antara keduanya menghasilkan pesan hilang atau duplikat. Matikan enable.auto.commit dan serahkan offset kepada send_offsets_to_transaction.
3. Meninggalkan transaksi menggantung setelah menangkap error. Partisi yang menyimpan transaksi terbuka menahan LSO dan membekukan pembacaan read_committed di belakangnya, sehingga lag konsumen lain naik tanpa penyebab yang kelihatan. Panggil abort_transaction pada jalur penanganan error, atau tutup producer dan biarkan sesi berikutnya meng-abort sisa transaksinya.
4. Menetapkan transactional.id berbeda pada setiap restart. Broker tidak dapat meng-fencing instance lama karena pengenalnya tidak pernah sama, dan transaksi yang menggantung tidak di-abort otomatis. Gunakan pengenal yang stabil untuk satu instance logis dalam aplikasi.

Ringkasan#

• Transaksi producer menjadikan pesan ke banyak partisi dan commit offset konsumen satu kesatuan atomik, diatur lewat init_transactions, begin_transaction, send_offsets_to_transaction, commit_transaction, dan abort_transaction.
• transactional.id yang stabil mengaktifkan fencing antar sesi, sehingga producer lama ditolak dan transaksi yang menggantung di-abort otomatis.
• isolation.level=read_committed menyembunyikan pesan transaksi yang di-abort dan berhenti sampai LSO, sedangkan read_uncommitted mengembalikan semuanya.
• Pola consume-transform-produce menuntut commit otomatis yang dimatikan dan commit offset lewat transaksi agar hasil transformasi serta posisi konsumen commit bersama-sama.

Referensi#

• Apache Kafka design: using transactions — Apache Software Foundation — prinsip transaksi producer, peran transactional.id, dan prasyarat pemrosesan tepat sekali.
• KafkaProducer — Apache Software Foundation — lima panggilan transaksi klien Java yang dipetakan ke kafka-python pada topik ini.
• Producer configs — Apache Software Foundation — konfigurasi transactional.id beserta kaitannya dengan pengiriman idempoten.
• Consumer configs — Apache Software Foundation — nilai isolation.level, perilaku read_committed, dan batas pembacaan sampai last stable offset.
• kafka-python KafkaProducer — kafka-python project — dokumentasi metode transaksi kafka-python yang digunakan pada demo.