Transactions

ไอเดียในหนึ่งประโยค

transaction ของ Kafka ให้ producer เขียนไปหลาย partition (และ commit consumer offset) แบบ atomic ทุกอย่างใน transaction ปรากฏพร้อมกันตอน commit หรือไม่ปรากฏเลยตอน abort

ตั้งค่า transactional producer

transaction ต่อยอดจาก idempotent producer คุณเพิ่ม transactionalId ที่เป็น id ที่เสถียรและให้ Kafka fence ตัว zombie ของ producer เชิงตรรกะเดียวกันออกไป พร้อมกับ idempotent: true และ maxInFlightRequests: 1 การ connect producer จะลงทะเบียน transactional id ให้เอง — ต่างจาก client ฝั่ง Java ตรงที่ KafkaJS ไม่มี initTransactions() แยกต่างหาก

import { Kafka } from 'kafkajs'

const kafka = new Kafka({ clientId: 'payments', brokers: ['localhost:9092'] })

// A transactional producer: a stable transactionalId is the key ingredient
const producer = kafka.producer({
  transactionalId: 'payments-tx-1',   // stable id enables transactions + zombie fencing
  idempotent: true,                    // required foundation
  maxInFlightRequests: 1,
})

await producer.connect()               // connecting registers the transactional id

วงจรชีวิตของ transaction

เริ่มหน่วยงานหนึ่งด้วย producer.transaction() send ทุกตัวบน txn ที่ได้คืนมาถูก buffer เข้า transaction txn.commit() ทำให้ทุกตัวปรากฏพร้อมกันแบบ atomic และ txn.abort() ทิ้งทั้งหมด

const txn = await producer.transaction()   // begin
try {
  await txn.send({ topic: 'accounts', messages: [{ key: debitKey, value: debit }] })
  await txn.send({ topic: 'accounts', messages: [{ key: creditKey, value: credit }] })
  await txn.commit()     // both writes appear together, or neither
} catch (e) {
  await txn.abort()      // roll back the whole unit
}

consumer ต้องอ่านแบบ committed

transaction มีความหมายก็ต่อเมื่อฝั่งอ่านเมิน data ที่ abort และที่ยัง in-flight ตั้ง consumer เป็น readUncommitted: false — คือ read_committed และเป็นค่า default ของ KafkaJS — เพื่อให้ consumer ไม่เห็น record จาก transaction ที่ abort หรือที่ยังเปิดอยู่

const consumer = kafka.consumer({
  groupId: 'ledger',
  readUncommitted: false,   // read_committed: skip aborted and in-flight records
})

flowchart LR
  init["connect() once (registers txn id)"] --> begin["producer.transaction()"]
  begin --> s1["txn.send to accounts-0"]
  begin --> s2["txn.send to accounts-2"]
  s1 --> commit["txn.commit(): all visible together"]
  s2 --> commit
  begin -.->|on error| abort["txn.abort(): nothing visible"]

transaction commit หลาย write แบบ atomic

KIP-890 ใน Kafka 4.x

Kafka 4.x finalize KIP-890 Transactions Server-Side Defense ที่เสริมความแข็งแรงให้ transactional protocol รับมือ edge case อย่าง transaction ที่ค้างและ zombie producer พอ finalize การอัปเกรด 4.0 protocol ใหม่ที่แข็งแรงกว่าก็ถูกเปิดใช้ ทำให้ transactional pipeline บน Kafka 4.x ทนทานกว่าเวอร์ชันเก่าโดยที่คุณไม่ต้องแก้ code ฝั่งตัวเอง