Offsets & Commits

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

consumer บันทึกว่าประมวลผลไปถึงไหนด้วยการ commit offset ต่อ partition Kafka เก็บ offset เหล่านี้ไว้ใน topic ภายในชื่อ __consumer_offsets และ จุดที่คุณ commit เทียบกับ จุดที่คุณ process เป็นตัวตัดสินว่าคุณจะทำ record หายหรือซ้ำได้

committed เทียบกับ current position

แต่ละ partition พก offset สองตัวมาด้วย:

Position offset ตัวถัดไปที่ consumer จะ fetch (ขยับไปเรื่อยเมื่อ consume)
Committed offset offset ตัวสุดท้ายที่ consumer บันทึกอย่างถาวรว่า “เสร็จแล้ว”

เมื่อ restart หรือ rebalance consumer จะกลับมาเริ่มจาก committed offset ดังนั้น commit จึงเป็นคำสัญญาว่า “ทุกอย่างก่อนหน้านี้ process แล้ว อย่าส่งมาอีก”

auto-commit สะดวกแต่เสี่ยง

โดยดีฟอลต์ KafkaJS จะ commit ให้อัตโนมัติ (autoCommit: true): โดย commit offset ที่ส่งไปแล้วเป็นระยะ ทุก ๆ autoCommitInterval มิลลิวินาที (หรือหลังครบ autoCommitThreshold message) วิธีนี้ง่าย แต่ commit ตามตารางเวลาโดยไม่สนว่าการประมวลผลของคุณสำเร็จหรือไม่ ถ้า crash หลัง auto-commit แต่ก่อน process จะทำให้ record เหล่านั้น หาย สำหรับงานที่สำคัญ ให้ตั้ง autoCommit: false แล้ว commit หลัง process:

const consumer = kafka.consumer({ groupId: 'order-processor' });
await consumer.connect();
await consumer.subscribe({ topic: 'orders' });

await consumer.run({
  autoCommit: false,                                    // take control of commits
  eachMessage: async ({ topic, partition, message }) => {
    await process(message);                             // do the work FIRST
    // then commit the NEXT offset to read (offset + 1) — at-least-once
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

manual commit ใน KafkaJS

เมื่อ autoCommit: false คุณ commit เองด้วย consumer.commitOffsets([...]) ซึ่งคืน Promise ให้ await ทุกครั้ง เพื่อให้แน่ใจว่า commit ถูก ack ก่อนไปต่อ — ตัวเลือกที่ปลอดภัยและช้ากว่านิดหน่อย เหมือน commit แบบ synchronous ถ้าไม่ await ก็ปล่อยให้ commit วิ่งเบื้องหลังเพื่อ throughput โดยยอมรับว่า error จะไม่ถูกจับตรงนั้น ไม่ว่าทางไหนคุณ commit offset ตัวถัดไป ที่ต้องอ่าน (offset + 1) เพราะนั่นคือจุดที่ consumer ควรกลับมาเริ่ม

จะเริ่มอ่านจากไหน และ replay ด้วย seek()

committed offset จะมีประโยชน์ก็ต่อเมื่อมีอยู่จริง ครั้งแรกที่ group อ่าน partition หนึ่ง หรือถ้า committed offset หมดอายุตาม retention ไปแล้ว flag fromBeginning บน subscribe() ของ KafkaJS จะเป็นตัวตัดสินว่าจะเริ่มตรงไหน:

fromBeginning: true เริ่มจากต้น partition แล้ว process history ที่เก็บไว้ทั้งหมดใหม่ (policy earliest)
fromBeginning: false เริ่มจากท้าย อ่านเฉพาะ record ที่เข้ามานับจากนี้ (ดีฟอลต์ latest)

คุณยัง override ตำแหน่งเองได้ด้วย consumer.seek() ย้าย partition ไปที่ offset ไหนก็ได้เพื่อ replay record เก่าหรือข้ามไปข้างหน้า:

// where to begin when there is no committed offset:
// fromBeginning: true = earliest (replay all history), false = latest (default)
await consumer.subscribe({ topic: 'orders', fromBeginning: true });

// Replay partition 0 of "orders" from offset 100 (call after run() has started)
consumer.seek({ topic: 'orders', partition: 0, offset: '100' }); // next fetch starts at offset 100
// use offset '0' to replay the whole partition from the start

flowchart LR
  poll["poll batch"] --> proc["process records"]
  proc --> commit["commitOffsets() after success"]
  commit --> poll
  proc -.->|crash before commit| replay["restart re-reads from last commit (duplicates)"]

Commit หลัง process ให้ผลเป็น at-least-once

ทำไมดีฟอลต์นี้จึงเป็น at-least-once

commit หลัง process แล้ว crash ระหว่างสองจุดนั้น หมายความว่า record เหล่านั้นจะถูกอ่านและ process ซ้ำตอน restart คือ at-least-once อาจซ้ำได้ ส่วน commit ก่อน process แล้ว crash หมายถึง record เหล่านั้นถูกข้ามไป คือ at-most-once อาจหายได้ at-least-once เป็นดีฟอลต์ที่ใช้กันทั่วไป จะยกระดับเป็น exactly-once จริง ๆ ต้องอาศัย idempotent processing หรือ transaction ซึ่ง module ถัดไปจะครอบคลุม

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

consumer commit offset เพื่อทำเครื่องหมายความคืบหน้า (เก็บใน __consumer_offsets) และกลับมาเริ่มจาก committed offset หลัง restart auto-commit (autoCommit: true) ง่ายแต่ commit ตามตารางเวลาและอาจทำ record หาย เพื่อความถูกต้องให้ตั้ง autoCommit: false แล้ว commit หลัง process ด้วย consumer.commitOffsets([...]) การ commit หลัง process ให้ผลเป็น at-least-once (อาจซ้ำ) คือดีฟอลต์ที่ต้องออกแบบเผื่อไว้ ใช้ fromBeginning ตัดสินว่า group ใหม่เอี่ยมจะเริ่มตรงไหน (earliest/latest) และใช้ seek() เพื่อ replay