Error Handling & Retries

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

code Kafka ที่ทนทานจะจัดทุก failure ลงหนึ่งในสามถัง retriable (ปล่อยให้ retry policy ของ KafkaJS retry ให้), abortable (roll transaction กลับแล้วลอง batch ใหม่) หรือ fatal (disconnect producer แล้วให้อีก instance เข้ามาแทน) และ dead-letter record ที่ไม่มีวันสำเร็จ

producer retry อัตโนมัติและมีกรอบ

KafkaJS retry error แบบ retriable เอง เช่น leader election หรือ network สะดุดชั่วคราว โดยคุมด้วย retry policy ของ client: จำนวน retries ที่จำกัด พร้อม exponential backoff (แต่ละรอบเริ่มที่ initialRetryTime แล้วโตขึ้นไปจนถึง maxRetryTime) คุณตั้งครั้งเดียวบน Kafka client และครอบคลุมทั้งการ produce, consume และ admin call

const kafka = new Kafka({
  clientId: 'orders-app',
  brokers: ['localhost:9092'],
  retry: {
    retries: 8,             // bounded automatic retries for retriable errors
    initialRetryTime: 300,  // first backoff in ms; grows exponentially
    maxRetryTime: 30000,    // cap on any single backoff
  },
})

const producer = kafka.producer({ idempotent: true })  // keeps those retries duplicate-free

เมื่อ producer เป็น idempotent: true retry เหล่านี้ไม่สร้าง duplicate ใน partition เลย คุณจึง retry ได้ดุดันโดยไม่ทำ stream พัง

abortable vs fatal ใน transaction

ภายใน transactional loop KafkaJS จะ throw เมื่อ operation ล้มเหลว และ flag retriable บน error ที่ throw ออกมาเป็นตัวตัดสินวิธีกู้คืน

retriable / abortable (e.retriable === true) transaction commit ไม่ได้แต่ producer ยังปกติดี เรียก txn.abort() แล้ว retry batch จาก offset ที่ commit ล่าสุด
fatal (e.retriable === false) มี producer ตัวใหม่เข้ามายึด transactionalId ของคุณ (zombie ถูก fence) หรือ idempotent sequence พังแล้ว producer ตัวนี้กู้คืนไม่ได้ ให้ disconnect แล้ว shut down เพื่อให้ instance ที่ปกติทำงานต่อ

const txn = await producer.transaction()
try {
  for (const message of batch.messages) {
    await txn.send({ topic: 'orders-enriched', messages: [{ key: message.key, value: transform(message) }] })
  }
  await txn.sendOffsets({ consumerGroupId: 'enricher', topics: offsetsFor(batch) })
  await txn.commit()
} catch (e) {
  await txn.abort()                  // ABORTABLE: roll back and retry the batch
  if (!e.retriable) {                // FATAL: a zombie was fenced — give up this instance
    await producer.disconnect()
    throw e
  }
}

poison message, dead-letter topic และ safety net

poison message คือ record ที่ consumer ไม่มีวันประมวลผลได้ payload พัง schema ที่ parse ไม่ได้ หรือ downstream ที่ล้มถาวร ถ้า retry ไปเรื่อย ๆ record ตัวนี้จะบล็อกทั้ง partition ไว้ วิธีแก้มาตรฐานคือ dead-letter topic (DLT) หลังลองล้มเหลว N ครั้ง ก็ produce record บวก context ของ error ไปยัง topic orders-dlq แยกต่างหาก commit ข้าม record นั้นไป แล้วปล่อยให้ flow หลักเดินต่อ คนหรือ job แยกค่อยมาตรวจ DLT ทีหลัง

try {
  await process(message)
} catch (e) {
  // Route the bad record aside instead of blocking the partition
  await dlqProducer.send({ topic: 'orders-dlq', messages: [{ key: message.key, value: message.value }] })
}
// commit past the record either way, so the partition keeps advancing

ใต้ทั้งหมดนี้ idempotent consumer คือ safety net เพราะ at-least-once หมายความว่า record ถูกส่งซ้ำได้หลังการล่มใด ๆ ให้ทำ logic ประมวลผลของคุณให้ทำซ้ำได้อย่างปลอดภัย เช่น UPSERT ที่ key ด้วย event id, conditional write หรือ dedupe table แล้ว retry, batch ที่ abort-แล้ว-retry หรือการส่งซ้ำหลัง producer ถูก fence ก็จะ converge ไปที่ state ที่ถูกต้องเดียวกัน

flowchart TB
  fail["failure during send or process"] --> retriable{"which kind?"}
  retriable -->|"transient"| retry["KafkaJS retries per its retry policy"]
  retriable -->|"abortable (e.retriable)"| abort["txn.abort() then retry batch"]
  retriable -->|"fatal (not retriable)"| fatal["disconnect producer and shut down"]
  retriable -->|"poison message"| dlq["send to dead-letter topic, commit past it"]

จัดแต่ละ failure retry, abort-แล้ว-retry, dead-letter หรือ shut down

🌊 สรุปหนึ่งบรรทัด

จัด failure ตามชนิด error ชั่วคราวจัดการด้วย retry อัตโนมัติจาก retry policy ของ KafkaJS (จำนวน retries ที่จำกัด พร้อม exponential backoff และ producer แบบ idempotent: true ทำให้ไม่ซ้ำ) error แบบ abortable (e.retriable) หมายถึง txn.abort() แล้ว retry batch ส่วนแบบ non-retriable เป็น fatal — disconnect แล้ว shut down route poison message ไป dead-letter topic และเก็บ idempotent consumer ไว้เป็น safety net เพื่อให้การส่งซ้ำใด ๆ ลงเอยที่ state เดียวกัน