Exactly-Once Processing

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

use case คลาสสิกของ exactly-once คือ consume-transform-produce อ่านจาก input topic แปลง แล้วเขียนไป output topic และ Kafka ทำให้ flow นี้เป็น exactly-once ด้วยการ commit input offset ภายใน transaction เดียวกัน กับ output record

ทำไม offset ต้องเข้าไปอยู่ใน transaction

ลองนึกถึงการอ่าน orders เติมข้อมูลแต่ละตัว แล้วเขียนไป orders-enriched ถ้าคุณ commit output ใน transaction แต่ commit input offset แยกต่างหาก การล่มระหว่างสองจังหวะจะทำให้ประมวลผลซ้ำ (output ซ้ำ) หรือข้าม (input หาย) วิธีแก้คือ commit offset ของ consumer ให้เป็นส่วนหนึ่งของ transaction ของ producer ด้วย txn.sendOffsets() ทีนี้ output record กับความคืบหน้าของ input จะ commit หรือ abort พร้อมกัน

const producer = kafka.producer({ transactionalId: 'enricher-1', idempotent: true, maxInFlightRequests: 1 })
const consumer = kafka.consumer({ groupId: 'enricher', readUncommitted: false })
await producer.connect()
await consumer.connect()
await consumer.subscribe({ topic: 'orders' })

await consumer.run({
  autoCommit: false,                               // offsets are committed inside the transaction
  eachBatch: async ({ batch }) => {
    const txn = await producer.transaction()       // begin
    try {
      for (const message of batch.messages) {
        await txn.send({ topic: 'orders-enriched', messages: [{ key: message.key, value: enrich(message.value) }] })
      }
      // Commit the input offsets in the SAME transaction as the output
      await txn.sendOffsets({
        consumerGroupId: 'enricher',
        topics: [{ topic: batch.topic, partitions: [{ partition: batch.partition, offset: (Number(batch.lastOffset()) + 1).toString() }] }],
      })
      await txn.commit()                           // outputs + offsets are atomic
    } catch (e) {
      await txn.abort()
    }
  },
})

จัดการ failure ให้ถูก

KafkaJS จะ throw ถ้า transactional operation ล้มเหลว และ error ที่ throw ออกมาจะพก flag retriable ที่ตัดสินวิธีกู้คืน เคสปกติคือ retriable: txn.abort() แล้วปล่อยให้ batch รันใหม่จาก offset ที่ commit ล่าสุด ส่วน error แบบ non-retriable (e.retriable === false) แปลว่ามี producer ตัวใหม่เข้ามายึด transactionalId ของคุณไปแล้ว คุณไปต่อไม่ได้ ให้ disconnect() instance นี้แล้วปล่อยให้ตัวที่ยังดีทำงานแทน

} catch (e) {
  await txn.abort()                  // roll back and retry the batch
  if (!e.retriable) {                // a zombie was fenced — this producer cannot recover
    await producer.disconnect()      // give up this instance
    throw e
  }
}

flowchart LR
  in["consume from orders"] --> begin["producer.transaction()"]
  begin --> transform["transform + txn.send to orders-enriched"]
  transform --> off["txn.sendOffsets(input offsets)"]
  off --> commit["txn.commit(): outputs + offsets atomic"]
  commit --> in

consume-transform-produce เป็นหน่วย atomic เดียว