ส่งซ้ำได้ ไม่ใช่ส่งครั้งเดียว
producer ส่ง batch ออกไป แล้ว ack ไม่กลับมา สายอาจขาดตอนคำขอกำลังไป หรือขาดตอน ack กำลังกลับ ทั้งสองกรณีหน้าตาเหมือนกันเป๊ะจากฝั่ง producer
มันจึงมีทางเลือกแค่สองทาง คือส่งใหม่แล้วเสี่ยงได้ของซ้ำ หรือไม่ส่งแล้วเสี่ยงของหาย ทางเลือกนี้ไม่ได้มาจากการออกแบบที่แย่ มันมาจากการที่ปลายทั้งสองข้างมองไม่เห็นกัน
บทนี้เปิดทางที่สาม ชื่อของมันคือ idempotent produceridempotent producerproducer ที่ส่งซ้ำได้โดยไม่เกิด record ซ้ำใน log เพราะ broker จำ id กับลำดับล่าสุดของมันไว้ ซึ่งเป็น producer ที่ส่งซ้ำได้เท่าที่อยากส่ง โดยมี broker เป็นคนดูออกเองว่า batch ก้อนนี้มันเคยเขียนไปแล้ว
ราคาที่จ่ายคือ struct เล็กๆ ห้าตัวต่อหนึ่ง producer ต่อหนึ่ง partition และมันกันซ้ำได้เฉพาะการส่งใหม่ที่เกิด ภายใน session เดียวกัน เท่านั้น
และของที่ได้มา ไม่ใช่ exactly-once ทั้งบทนี้ครึ่งหนึ่งพูดเรื่องเส้นที่มันไปไม่ถึง
crate ของคอร์สอยู่ที่ scripts/log-broker/ และชื่อ binary คือ klog
บทนี้เป็นเจ้าของ scripts/log-broker/src/idem.rs ทั้ง file
บล็อก code กับบล็อกผลรันทุกอันในบทเรียนนี้ถูกคัดมาจาก scripts/log-broker/src/idem.rs
และจาก scripts/log-broker/expected/07-idem.txt ทีละ byte
โดยมีเทสต์เทียบไว้ ถ้าแก้ในบทเรียนอย่างเดียวโดยไม่แก้ที่ต้นทาง เทสต์จะแดงทันที
ผลรันทุกบรรทัดในบทนี้มาจากคำสั่งเดียว คือ
klog idem-demo --pid 1000 --seq 7..9 --resend --now-ms 1700000000000
ซึ่งเทสต์รันซ้ำแล้วเทียบผลทีละ byte ทุกครั้ง
สามความหมายของการส่ง และคำที่ข้อที่สามใช้ไม่เหมือนกัน
หัวข้อที่มีชื่อว่า “สามความหมายของการส่ง และคำที่ข้อที่สามใช้ไม่เหมือนกัน”หัวข้อนี้ยกนิยามสามข้อของ Kafka มาวางเรียงกัน แล้วชี้คำที่ข้อที่สามใช้ต่างจากสองข้อแรก
เอกสารออกแบบของ Kafka นิยามการส่งไว้สามแบบ และ demo ของบทนี้พิมพ์ทั้งสามแบบออกมาเป็นอย่างแรก
== สามความหมายของการส่ง == at most once ข้อความหายได้ แต่ไม่เคยถูกส่งซ้ำ at least once ข้อความไม่เคยหาย แต่ถูกส่งซ้ำได้ exactly once ข้อความถูก **ประมวลผล** ครั้งเดียวเท่านั้นสองข้อแรกพูดถึงการ *ส่ง* ข้อที่สามพูดถึงการ *ประมวลผล* · ความต่างนี้ไม่ใช่เรื่องถ้อยคำคำในเอกสารต้นทางเป็นอย่างนี้ At most once คือ “Messages may be lost but are never redelivered” At least once คือ “Messages are never lost but may be redelivered” ส่วน Exactly once คือ “Each message is processed once and only once”
สองข้อแรกใช้คำว่า lost กับ redelivered ซึ่งเป็นคำของการ ส่ง ข้อที่สามเปลี่ยนไปใช้คำว่า processed ซึ่งเป็นคำของการ ประมวลผล
ความไม่สมมาตรตรงนี้ไม่ใช่ความสะเพร่าของคนเขียนเอกสาร มันคือคำสารภาพ ไม่มีใครส่งข้อความบนสายให้ถึงครั้งเดียวจริงๆ ได้ สิ่งที่ทำได้คือทำให้ผลของการรับซ้ำเท่ากับรับครั้งเดียว
เจ้าของแต่ละข้อจึงเป็นคนละองค์ประกอบกัน
- at most once เป็นของ producer producer ที่ไม่ลองส่งใหม่เลยได้ข้อนี้ทันที และเสียข้อความไปเงียบๆ
- at least once เป็นของ producer บวก broker producer ลองส่งใหม่ ส่วน broker รับประกันว่าสิ่งที่ ack ไปแล้วอยู่บน disk
- exactly once ไม่ได้เป็นของใครคนเดียว มันต้องให้ผู้อ่านกับผู้เขียนตกลงกันข้ามระบบ ซึ่งบทนี้ไม่ได้สร้าง
เอกสารหน้าเดียวกันเตือนเรื่องนี้เอง ว่าหลายระบบอ้างว่าให้ exactly-once แต่ “it is important to read the fine print, because sometimes these claims are misleading” แล้วไล่กรณีที่ข้ออ้างนั้นพัง คือตอนผู้อ่านหรือ producer ล้ม ตอนมีผู้อ่านหลายราย และตอนของบน disk หาย
คอร์ส Distributed Systems สอนการตัดสินด้วยเสียงข้างมาก ทั้ง quorum แบบ R บวก W มากกว่า N และการซ่อมตอนอ่าน กลไกพวกนั้นตัดสินโดยการ นับ สำเนาหลายชุด
การกันซ้ำในบทนี้ไม่ได้นับอะไรเลย มันคือตัวนับ ต่อผู้เขียนหนึ่งราย ที่ broker เจ้าของ partition ถือไว้ฝ่ายเดียว ไม่มีการโหวต และไม่มีสำเนาที่สองเข้ามาเกี่ยว
field สามช่องที่รออยู่ในหัว batch มาตั้งแต่บทแรก
หัวข้อที่มีชื่อว่า “field สามช่องที่รออยู่ในหัว batch มาตั้งแต่บทแรก”หัวข้อนี้บอกว่าการกันซ้ำใช้ช่องไหนในหัว batch และทำไมช่องพวกนั้นถึงมีมาก่อนแล้ว
บทที่ 1 กางหัว batch ออกมาทีละช่อง
และในตารางนั้นมีสามช่องที่มีค่าเป็นลบหนึ่งทั้งหมด คือ producer_id producer_epoch กับ base_sequence
ลบหนึ่งในสามช่องนั้นแปลว่ายังไม่มีใครใช้ บทนี้คือบทที่เริ่มใช้มัน
การกันซ้ำจึงไม่ใช่ feature ที่แปะเข้ามาทีหลัง มันเป็นการตัดสินใจเรื่องรูปแบบของข้อมูล ที่ถูกตัดสินไปตั้งแต่ตอนออกแบบหัว batch ถ้าตอนนั้นไม่เผื่อช่องไว้ ตอนนี้จะเติมไม่ได้เลย
producer idproducer idเลขประจำตัวที่ broker แจกให้ producer · เป็นกุญแจของหน้าต่างกันซ้ำฝั่ง broker คือเลขประจำตัวที่ broker แจกให้ producer ตอนเริ่ม session KIP-98 เขียนไว้ว่า “Every new producer will be assigned a unique PID during initialization” และย้ำว่าการแจกนี้ “is completely transparent to users and is never exposed by clients”
sequence numbersequence numberเลขลำดับต่อ producer ต่อ partition ที่เพิ่มทีละ 1 · broker ใช้ตัดสินว่า batch นี้ซ้ำ ขาดช่วง หรือรับได้ คือเลขลำดับต่อหนึ่ง producer ต่อหนึ่ง partition KIP เดียวกันเขียนว่า “For a given PID, sequence numbers will start from zero and be monotonically increasing, with one sequence number per topic partition produced to”
ประโยคหลังมีคำสำคัญอยู่ที่ per topic partition ตัวนับไม่ได้มีตัวเดียวต่อ producer producer หนึ่งรายที่เขียนลงสาม partition ถือตัวนับสามตัวแยกกัน
ส่วน producer_epoch เป็นเลขรุ่นของ session ซึ่งบทนี้ใช้เป็นตัวประกอบของการเทียบเท่านั้น
kaen-log ไม่มีกลไกกันตัวเก่าออกด้วย epoch เหมือน Kafka และไม่ได้อ้างว่ามี
สี่ตัวเลขต่อหนึ่ง batch ที่ broker จำเอาไว้
หัวข้อที่มีชื่อว่า “สี่ตัวเลขต่อหนึ่ง batch ที่ broker จำเอาไว้”หัวข้อนี้กางสิ่งที่ broker เก็บต่อหนึ่ง batch แล้วชี้ว่าสองค่าในนั้นคำนวณเอา ไม่ได้เก็บ
สิ่งที่ broker จำไว้ต่อหนึ่ง batch ไม่ใช่ตัว batch มันคือตัวเลขสี่ตัว
/// สี่ตัวเลขต่อหนึ่ง batch ที่เก็บไว้ · ตรงกับ `BatchMetadata` ของ Kafka/// ซึ่งที่ tag 4.1.0 ประกาศไว้เป็น `public class` ที่มี `public final` สี่ช่องนี้พอดี/// (บน trunk มันถูกเขียนใหม่เป็น `record` แล้ว แต่บทนี้อ้างรุ่น 4.1.0 ทุกที่)#[derive(Clone, Copy, Debug, PartialEq, Eq)]pub struct BatchMetadata { pub last_seq: i32, pub last_offset: i64, pub offset_delta: i32, pub timestamp: i64,}
impl BatchMetadata { pub fn first_seq(&self) -> i32 { self.last_seq.wrapping_sub(self.offset_delta) }
pub fn first_offset(&self) -> i64 { self.last_offset - i64::from(self.offset_delta) }}first_seq กับ first_offset ไม่ได้ถูกเก็บ ทั้งคู่คำนวณกลับจาก offset_delta
ซึ่งเป็นค่าเดียวกับที่บทที่ 1 บอกว่าเท่ากับจำนวน record ในหนึ่ง batch ลบหนึ่ง
Kafka ทำเหมือนกันทีละบรรทัด BatchMetadata.java ของ tag 4.1.0 ประกาศตัวเองเป็น
public class ที่มี public final สี่ช่องนี้พอดี แล้วมี firstSeq() กับ firstOffset()
ที่คำนวณกลับจาก offsetDelta เหมือนกัน
รูปของมันเปลี่ยนไปแล้วบน trunk ที่นั่นมันถูกเขียนใหม่เป็น record ซึ่งได้ตัวอ่านทั้งสี่ช่องมาฟรี
บทนี้อ้างรุ่น 4.1.0 ทุกที่ จึงเรียกมันตามรูปที่ tag นั้นมีจริง ไม่ใช่ตามรูปบน trunk
จำนวนของที่เก็บจึงเป็นค่าคงที่ ไม่ผูกกับขนาดของ batch ที่ producer ส่งมา batch ที่มี record หนึ่งตัวกับ batch ที่มีพันตัว กินที่ในสถานะของ broker เท่ากันเป๊ะ
สามคำตัดสิน และตัวกลางที่คนคาดไม่ถึง
หัวข้อที่มีชื่อว่า “สามคำตัดสิน และตัวกลางที่คนคาดไม่ถึง”หัวข้อนี้กาง enum ของคำตัดสินทั้งสาม แล้วอยู่กับตัวกลางที่ตอบว่าสำเร็จทั้งที่เป็นของซ้ำ
broker ตัดสิน batch ที่เข้ามาได้สามแบบ และมันตัดสิน ก่อน ที่จะเขียนอะไรลง disk
/// สามคำตัดสิน · ตัวที่คนคาดไม่ถึงคือตัวกลาง#[derive(Clone, Copy, Debug, PartialEq, Eq)]pub enum Verdict { /// `first_seq` เท่ากับตัวที่คาดหวังพอดี Append, /// **ตอบว่าสำเร็จ พร้อม offset ชุดเดิม** ไม่ใช่ตอบว่าผิดพลาด Duplicate { first_offset: i64, last_offset: i64, timestamp: i64 }, /// มี **ช่องโหว่** · การเขียนหายไปหนึ่งครั้ง ไม่ใช่ซ้ำ /// การกันซ้ำจึงเปลี่ยนการสูญหายแบบเงียบๆ ให้เป็นข้อผิดพลาดที่ดังออกมา OutOfOrderSequenceNumber { expected: i32, got: i32 },}ตัวที่คนคาดไม่ถึงคือ Duplicate เพราะสัญชาตญาณบอกว่าของซ้ำควรได้คำตอบว่าผิดพลาด
แต่ broker ตอบว่าสำเร็จ แล้วส่ง offset ชุดเดิมกลับไปให้
เหตุผลอยู่ที่คำถามว่า producer ที่ได้คำตอบนั้นจะทำอะไรต่อ มันส่งซ้ำเพราะมันไม่รู้ผลของครั้งแรก คำตอบที่มันต้องการคือผลของครั้งแรก ไม่ใช่ป้ายว่ามันทำผิด
UnifiedLog.java ของ Kafka รุ่น 4.1.0 เขียนสาขานี้ไว้ตรงตัว เมื่อ result.maybeDuplicate
มีค่า มันเซ็ต setFirstOffset setLastOffset และ setLogAppendTime จากของเก่า
แล้วสาขา else ที่เรียก localLog.append(...) ถูกข้ามไปทั้งก้อน
ตัวเทียบเองก็เป็นแบบช่วงตรงเป๊ะ ไม่ใช่แบบคลุมเครือ
/// การเทียบเป็นแบบ **ช่วงตรงเป๊ะ** · ทั้ง `first_seq` และ `last_seq` ต้องตรงกับ/// batch ที่เก็บไว้ และ epoch ต้องตรงด้วยpub fn judge(entry: &ProducerStateEntry, epoch: i16, first: i32, last: i32) -> Verdict { let exact_match = entry .batches .iter() .find(|b| b.first_seq() == first && b.last_seq == last) .filter(|_| epoch == entry.producer_epoch); if let Some(b) = exact_match { return Verdict::Duplicate { first_offset: b.first_offset(), last_offset: b.last_offset, timestamp: b.timestamp, }; } let expected = entry.expected_first_seq(); if first == expected { Verdict::Append } else { Verdict::OutOfOrderSequenceNumber { expected, got: first } }}ทั้งปลายซ้ายและปลายขวาของช่วงต้องตรง และ epoch ต้องตรงด้วย
findDuplicateBatch ของ Kafka ตัดที่ epoch ก่อน แล้วค่อยหา batch ที่มีช่วง sequence ตรงเป๊ะ
ผลที่ตามมาคือ batch ที่ซ้อนทับกันครึ่งเดียวไม่ใช่ของซ้ำ
producer ที่รวมสอง batch เป็นก้อนเดียวแล้วส่งใหม่ จะไม่ได้คำตัดสิน Duplicate
ทางเดินของการเขียนจึงตัดสินก่อน แล้วค่อยเขียน
/// ตัดสินก่อน แล้วค่อยเขียน · batch ที่ซ้ำ **ไม่ถูกเขียนลง disk เลย**pub fn append_idempotent( partition: &Partition, state: &mut ProducerState, producer_id: i64, producer_epoch: i16, first_seq: i32, records: Vec<Record>, now_ms: i64,) -> io::Result<IdempotentAppend> { let count = i32::try_from(records.len()).map_err(|_| bad("batch เดียวเกิน i32 record"))?; let last_seq = first_seq + count - 1; let entry = state.entry_mut(producer_id, producer_epoch); let verdict = judge(entry, producer_epoch, first_seq, last_seq);
match verdict { Verdict::Duplicate { first_offset, .. } => { Ok(IdempotentAppend { verdict, base_offset: first_offset }) } Verdict::OutOfOrderSequenceNumber { .. } => { Ok(IdempotentAppend { verdict, base_offset: -1 }) }สองสาขาแรกคืนค่าออกไปเลยโดยไม่แตะ log สาขาที่สามที่ถูกตัดออกไปจากบล็อกข้างบน คือสาขาเดียวที่ประกอบ batch แล้วเรียกทางเขียนของบทที่ 3
sequenceDiagram
autonumber
participant P as producer · pid 1000
participant B as broker · orders-0
P->>B: ส่ง batch · seq 7..9
B->>B: ตัดสิน APPEND แล้วต่อท้าย log
B--xP: ack หายกลางทาง
Note over P: ไม่รู้ว่าเขียนสำเร็จหรือไม่ · จึงส่งใหม่
P->>B: ส่ง batch เดิมซ้ำ · seq 7..9
B->>B: เจอช่วงตรงเป๊ะในหน้าต่างห้าก้อน
B-->>P: สำเร็จ · offset ชุดเดิม
Note over B: ไม่เขียนอะไรลง disk · ปลาย log ไม่ขยับ
คำบรรยายภาพ: ภาพนี้เป็นตัวอย่าง ไม่ใช่ผลรัน · การส่งซ้ำครั้งที่สองเดินเข้าเส้นทางเดียวกันทั้งหมด ต่างกันแค่ตรงที่ broker เจอช่วง sequence เดิมในหน้าต่างของตัวเอง จึงตอบด้วยผลของครั้งแรกแทนที่จะเขียนใหม่
ผลรันของจริงพูดข้อเดียวกัน
== ack หาย · producer ส่ง seq 7..9 ซ้ำ ==คำตัดสิน DUPLICATE · offset เดิม 7..9base_offset 7log end offset 10 (ไม่ขยับ)**ของซ้ำไม่ใช่ข้อผิดพลาด** broker ตอบว่าสำเร็จ พร้อม offset ชุดเดิมและไม่เขียนอะไรลง disk เลย · ทาง append ถูกข้ามทั้งเส้นบรรทัดที่พิสูจน์ข้ออ้างคือบรรทัด log end offset ที่ค้างอยู่ที่ค่าเดิม ถ้า batch ที่ซ้ำถูกเขียนลงไปจริง ตัวเลขนั้นจะขยับ และ demo จะจบด้วยข้อผิดพลาดทันที
หน้าต่างห้า batch และเลข 5 ที่โผล่สองที่แต่เป็นตัวเดียวกัน
หัวข้อที่มีชื่อว่า “หน้าต่างห้า batch และเลข 5 ที่โผล่สองที่แต่เป็นตัวเดียวกัน”หัวข้อนี้บอกว่า broker จำ batch ไว้ได้กี่ก้อน อะไรถูกดันออก และเลขนั้นผูกกับ config ตัวไหน
หน้าต่างกันซ้ำมีขนาดตายตัว และเลขของมันมาจาก Kafka ตรงๆ
/// **ห้า batch ไม่ใช่ห้า record** · ตรงกับ `NUM_BATCHES_TO_RETAIN = 5` ของ Kafka/// และเป็นเลข 5 ตัวเดียวกับ `max.in.flight.requests.per.connection` ที่ต้องไม่เกิน 5pub const NUM_BATCHES_TO_RETAIN: usize = 5;ProducerStateEntry.java ของรุ่น 4.1.0 ประกาศ public static final int NUM_BATCHES_TO_RETAIN = 5;
และ javadoc ของมันบอกวิธีจัดคิวไว้ด้วย ว่า sequence ต่ำสุดอยู่หัวคิว สูงสุดอยู่ท้ายคิว
แล้ว “When the queue is at capacity, we remove the first element to make space for the incoming batch”
ห้าก้อนนั้นคือห้า batch ไม่ใช่ห้า record batch หนึ่งก้อนอาจมี record กี่ตัวก็ได้ คนที่อ่านเป็นห้า record จะประเมินขอบของการกันซ้ำต่ำกว่าความจริงมาก
kaen-log ทำแบบเดียวกันด้วยคิวสองปลาย
/// คิวเต็มแล้วดันตัวหน้าออก · batch ที่ถูกดันออกไม่ถูกกันซ้ำอีกต่อไป pub fn record(&mut self, meta: BatchMetadata) { if self.batches.len() == NUM_BATCHES_TO_RETAIN { self.batches.pop_front(); } self.last_timestamp = meta.timestamp; self.batches.push_back(meta); }
/// sequence ตัวถัดไปที่คาดหวัง · producer ที่ยังไม่เคยเขียนอะไรเลยเริ่มที่ 0 pub fn expected_first_seq(&self) -> i32 { self.batches.back().map_or(0, |b| b.last_seq.wrapping_add(1)) }demo เดินหน้าต่างให้ดูทีละ batch จนเต็มพอดี
== หน้าต่างกันซ้ำ · 5 batch ต่อหนึ่ง producer ต่อหนึ่ง partition ==producer id 1000 · producer epoch 0 seq ที่ส่ง คำตัดสิน base_offset หน้าต่างหลังจากนั้น 0..1 APPEND 0 [0..1] 2..3 APPEND 2 [0..1 2..3] 4..5 APPEND 4 [0..1 2..3 4..5] 6..6 APPEND 6 [0..1 2..3 4..5 6..6] 7..9 APPEND 7 [0..1 2..3 4..5 6..6 7..9]log end offset 10แถวสุดท้ายมี batch อยู่ห้าก้อนพอดี ก้อนถัดไปที่เข้ามาจะดันก้อนหัวคิวออก
และหลังจากนั้น การส่งซ้ำของก้อนที่ถูกดันออกจะไม่ได้คำตัดสิน Duplicate อีก
== หน้าต่างล้น · batch ที่ถูกดันออกไม่ถูกกันซ้ำอีก ==หน้าต่างตอนนี้ [10..10 11..11 12..12 13..13 14..14]ส่ง seq 7..9 ซ้ำอีกครั้ง → OUT_OF_ORDER_SEQUENCE_NUMBER · คาดว่า 15 ได้ 7เลข 5 ตัวนี้เป็นตัวเดียวกับเพดานของ max.in.flight.requests.per.connectionส่งค้างไว้เกินห้า แล้ว batch เก่าจะหลุดหน้าต่างก่อนที่ ack จะกลับมาช่วง 7..9 ที่เมื่อกี้ได้คำตัดสินว่าซ้ำ ตอนนี้ได้คำตัดสินว่าลำดับผิด
สิ่งที่เปลี่ยนไม่ใช่ batch มันคือสิ่งที่ broker ยังจำได้
เลข 5 ที่โผล่ในสองที่เป็นตัวเดียวกัน เอกสารของ ProducerConfig เขียนไว้เองว่า
การเปิด idempotence บังคับให้ max.in.flight.requests.per.connection ไม่เกินห้า
“because broker only retains at most 5 batches for each producer”
ประโยคที่ว่า producer แบบ idempotent ส่งซ้ำได้โดยไม่เกิดของซ้ำ เป็นจริงในขอบของหน้าต่างนี้เท่านั้น
ถ้าปล่อยให้มีคำขอค้างอยู่บนสายเกินห้าก้อน แล้ว ack ของก้อนแรกเพิ่งกลับมาช้า batch ก้อนนั้นอาจหลุดหน้าต่างไปแล้วตอนที่ producer ตัดสินใจส่งซ้ำ
นั่นคือเหตุผลที่ Kafka บังคับเพดานไว้ที่ห้า ไม่ใช่เพราะห้าเป็นเลขสวย และเป็นตัวอย่างของ config ที่ผูกกับโครงสร้างข้อมูลฝั่ง broker โดยตรง
ช่องโหว่แปลว่าการเขียนหายไป ไม่ใช่ของซ้ำ
หัวข้อที่มีชื่อว่า “ช่องโหว่แปลว่าการเขียนหายไป ไม่ใช่ของซ้ำ”หัวข้อนี้อ่านคำตัดสินตัวที่สาม แล้วเถียงว่ามันอาจมีค่ากว่าการกันซ้ำเอง
คำตัดสินตัวที่สามเกิดเมื่อ first_seq ที่เข้ามาไม่ตรงกับตัวที่ broker คาดไว้ และไม่ตรงกับของเก่าด้วย
== ช่องโหว่ · ข้ามไปสามลำดับ ==คำตัดสิน OUT_OF_ORDER_SEQUENCE_NUMBER · คาดว่า 10 ได้ 13log end offset 10 (ไม่ขยับ)ช่องโหว่แปลว่ามีการเขียนที่ **หายไป** ไม่ใช่ที่ซ้ำการกันซ้ำจึงเปลี่ยนการสูญหายแบบเงียบๆ ให้เป็นข้อผิดพลาดที่ดังออกมาซึ่งอาจเป็นของที่มีค่ากว่าการกันซ้ำเสียอีกความหมายของมันคือ มีการเขียนหายไป ไม่ใช่มีการเขียนซ้ำ sequence เป็นเลขที่เพิ่มทีละหนึ่ง ช่องว่างที่เดินล้ำไปข้างหน้าจึงแปลว่าของที่ควรมาก่อนยังไม่เคยมาถึง
Errors.java ของรุ่น 4.1.0 มีสองรหัสที่เกี่ยวกับเรื่องนี้อยู่ติดกัน
OUT_OF_ORDER_SEQUENCE_NUMBER(45, "The broker received an out of order sequence number.")DUPLICATE_SEQUENCE_NUMBER(46, "The broker received a duplicate sequence number.")
ตัวที่ 45 คือตัวที่ broker โยนจริง ProducerAppendInfo.java โยน OutOfOrderSequenceException
เมื่อ inSequence(currentLastSeq, appendFirstSeq) เป็นเท็จ และ inSequence เองก็สั้นมาก
มันแค่ถามว่า nextSeq == lastSeq + 1L หรือเป็นกรณีวนกลับที่ปลายของ Integer.MAX_VALUE
ตัวที่ 46 เป็นอีกเรื่อง การไล่หาทางเดินที่โยนมันฝั่ง broker ไม่พบเลยใน ProducerAppendInfo.java
ProducerStateManager.java และ UnifiedLog.java ของรุ่น 4.1.0
ฝั่งที่จัดการมันจริงคือ client ใน Sender.java ซึ่งตอบผู้ใช้ว่าสำเร็จ
แล้วเขียนคอมเมนต์กำกับไว้เองว่า “The only thing we can do is to return success to the user and not
return a valid offset and timestamp”
ข้อนี้กลับด้านสิ่งที่คนคาดไว้ทั้งหมด ของซ้ำได้คำตอบว่าสำเร็จ ส่วนของที่หายได้คำตอบว่าผิดพลาด การกันซ้ำจึงเปลี่ยนการสูญหายแบบเงียบๆ ให้ดังออกมา ซึ่งเป็นของที่หาได้ยากกว่าการกันซ้ำเสียอีก
การรีเซ็ต sequence เพื่อให้ producer เดินต่อจากช่องโหว่ได้ เท่ากับยอมรับว่าของที่หายไปหายจริง ข้อนี้เป็นการอนุมานจากสองที่ที่ยืนยันได้ ไม่ใช่ประโยคที่ Kafka เขียนไว้เอง
สร้างสถานะขึ้นใหม่ตอนเปิด และขอบที่การกวาดไปไม่ถึง
หัวข้อที่มีชื่อว่า “สร้างสถานะขึ้นใหม่ตอนเปิด และขอบที่การกวาดไปไม่ถึง”หัวข้อนี้บอกว่าสถานะกันซ้ำรอด restart ของ broker ได้อย่างไร และมันเห็นแค่ไหน
สถานะที่บทนี้พูดถึงทั้งหมดอยู่ในหน่วยความจำ ไม่มี file แยกที่เก็บมันไว้ คำถามจึงตามมาทันที ว่า broker ที่เพิ่งเปิดใหม่รู้ได้อย่างไรว่า producer เดินมาถึง sequence ไหนแล้ว
คำตอบคือมันอ่านกลับจาก log ที่มันเพิ่งเปิด
/// สร้างสถานะขึ้นใหม่จากการกวาด **ท่อนสุดท้าย** ตอนเปิด////// kaen-log ไม่มี file `.snapshot` แบบ Kafka · การกวาดนี้คือเหตุผลเดียวที่ข้ออ้าง/// หลักของบทนี้รอดข้ามการ restart ของ broker · และมันเห็นแค่สิ่งที่ท่อนสุดท้าย/// ยังถืออยู่ producer ที่ batch ของมัน roll ตกไปท่อนเก่าแล้วจึงไม่ถูกกันซ้ำอีกpub fn rebuild_producer_state(log: &Log) -> io::Result<ProducerState> { let mut state = ProducerState::default(); let segment = log.active_segment(); for (_, batch) in segment.read_all_batches()? { let header = batch.header; if header.producer_id < 0 { continue; } let entry = state.entry_mut(header.producer_id, header.producer_epoch); entry.producer_epoch = header.producer_epoch; entry.record(BatchMetadata { last_seq: header.base_sequence.wrapping_add(header.last_offset_delta), last_offset: wire_i64(header.last_offset()?), offset_delta: header.last_offset_delta, timestamp: header.max_timestamp, }); } Ok(state)}ทุกค่าที่มันต้องใช้อยู่ในหัว batch อยู่แล้ว การกวาดจึงไม่ได้ใช้เนื้อของ record เลยสักตัว
last_seq ได้จาก base_sequence บวก last_offset_delta ซึ่งเป็นสองช่องที่อยู่ในหัว batch ทั้งคู่
บทที่ 6 กวาด log ด้วยการเขียน batch ขึ้นใหม่ ถ้าการเขียนใหม่นั้นรีเซ็ต producer_id
producer_epoch และ base_sequence กลับเป็นลบหนึ่ง การกวาดข้างบนจะไม่เจออะไรเลย
crate จึงมีตัวสร้าง batch สองตัวแยกกัน และตัวที่ compaction ใช้เป็นตัวที่รักษาสามช่องนั้นไว้
เอกสารของ DefaultRecordBatch ฝั่ง Kafka ให้เหตุผลเดียวกัน ว่าต้องรักษา sequence ตัวแรกกับตัวสุดท้ายไว้
เพื่อให้สร้างสถานะของ producer ขึ้นใหม่ได้ตอนอ่าน log กลับ
ข้อนี้พังตอนรัน ไม่ใช่ตอนอ่านบนกระดาษ
ผลรันยืนยันว่าการกวาดใช้ได้จริง
== broker restart · สร้างสถานะขึ้นใหม่จากการกวาดท่อนสุดท้าย ==producer id ที่กวาดเจอ [1000]หน้าต่างที่กวาดได้ [10..10 11..11 12..12 13..13 14..14]ส่ง seq 14..14 ซ้ำหลัง restart → DUPLICATE · offset เดิม 14..14หน้าต่างที่กวาดได้ตรงกับหน้าต่างก่อน restart ทุกช่อง และการส่งซ้ำหลัง restart ยังถูกกัน
สามขอบที่การกันซ้ำไปไม่ถึง
หัวข้อที่มีชื่อว่า “สามขอบที่การกันซ้ำไปไม่ถึง”หัวข้อนี้ไล่ข้อจำกัดสามข้อของสิ่งที่เพิ่งสร้างเสร็จ โดยสองข้อแรกมีผลรันกำกับ
ข้อจำกัดสามข้อข้างล่างไม่ใช่รายการสิ่งที่ยังทำไม่เสร็จ ทั้งสามข้อเป็นขอบของกลไกนี้เอง
ข้อแรก process ที่เริ่มใหม่ได้ producer id ตัวใหม่ เมื่อ id เปลี่ยน sequence ก็เริ่มนับใหม่จากศูนย์ batch ที่มี payload เดิมเป๊ะจึงเป็นของใหม่ในสายตาของ broker
== สองขอบที่การกันซ้ำไปไม่ถึง ==1. process ที่เริ่มใหม่ได้ producer id ตัวใหม่ ส่ง payload ชุดเดิมด้วย pid 1001 seq 0 → APPEND base_offset 15 · record ชุดเดิมถูกเขียนลง log เป็นของใหม่ทั้งหมด การกันซ้ำจึงครอบแค่ session เดียว ไม่ใช่ครอบ payloadการกันซ้ำจึงครอบ session ไม่ได้ครอบเนื้อของข้อความ
เอกสารของ ProducerConfig พูดข้อนี้กลับด้าน ว่าถ้าไม่ตั้ง transactional.id
“then the producer is limited to idempotent delivery” ส่วนสิ่งที่ข้าม session ได้คือ transaction
ข้อที่สอง การกวาดตอนเปิดเห็นแค่สิ่งที่ท่อนสุดท้ายยังถืออยู่ producer ที่ batch ของมันถูก roll ตกไปท่อนเก่าแล้ว จะไม่ถูกกันซ้ำอีกหลัง restart
2. การกวาดตอนเปิดเห็นแค่สิ่งที่ **ท่อนสุดท้าย** ยังถืออยู่ เขียน 8 batch ลง partition ที่ segment.bytes = 256 · จำนวนท่อน 8 หน้าต่างก่อน restart [3..3 4..4 5..5 6..6 7..7] หน้าต่างหลัง restart [7..7] แคบลงจาก 5 เหลือ 1 · producer ที่ batch ของมัน roll ตกไปท่อนเก่าแล้ว จึงไม่ถูกกันซ้ำอีกหลัง restart แม้ก่อนหน้านั้นจะถูกกันข้อนี้อ่านง่ายจากตัวเลขสองบรรทัด หน้าต่างก่อน restart เต็มห้าก้อน หลัง restart เหลือก้อนเดียว demo ตั้งขนาดท่อนไว้เล็กมากเพื่อบังคับให้เกิด และมันจบด้วยข้อผิดพลาดถ้าหน้าต่างไม่แคบลงจริง
Kafka ไม่มีปัญหาข้อนี้ในรูปเดียวกัน เพราะมัน snapshot สถานะของ producer ลง disk แยกไว้ แต่มันมีขอบของตัวเองคนละแบบ คือสถานะของ producer หมดอายุตามเวลา
ข้อที่สาม exactly-once ไม่เคยเป็นสมบัติของการส่ง มันเป็นสมบัติของ ผลลัพธ์ ที่ปลายทาง และปลายทางนั้นต้องกันซ้ำเป็นด้วยตัวมันเอง
การตัดบัตร การส่ง SMS และการยิงคำสั่งเข้าระบบภายนอก ไม่มีอะไรในบทนี้คุ้มครองมัน exactly-onceexactly-onceข้ออ้างที่เป็นจริงเฉพาะภายในขอบของระบบเดียว ประกอบขึ้นจากการกันซ้ำกับ transaction ไม่ใช่การส่งครั้งเดียวจริงๆ บนสาย จึงเป็นข้ออ้างที่เป็นจริงภายในขอบของระบบเดียวเท่านั้น
เอกสารออกแบบของ Kafka ไม่ได้เขียนประโยคนี้ด้วยถ้อยคำแบบนี้ ประโยคที่ใกล้ที่สุดที่มันเขียนคือ “Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the primitives which makes implementing this feasible”
และมันเสนอทางออกที่ง่ายกว่าไว้ในหน้าเดียวกัน ว่าให้ผู้อ่านเก็บตำแหน่งของตัวเองไว้ที่เดียวกับผลลัพธ์ เพราะ “many of the output systems a consumer might want to write to will not support a two-phase commit”
transaction บทนี้วาดไว้ ไม่ได้สร้าง
หัวข้อที่มีชื่อว่า “transaction บทนี้วาดไว้ ไม่ได้สร้าง”หัวข้อนี้อธิบายและวาดสิ่งที่ transaction ต้องมี แล้วประกาศว่า kaen-log ไม่ได้สร้างมันสักชิ้น
ประโยคต่อไปนี้ตรงตัว kaen-log ไม่มี transaction ทุกอย่างในหัวข้อนี้ถูกอธิบายและวาดไว้เฉยๆ ไม่มีบรรทัดไหนใน crate ที่ทำมัน และไม่มีบทไหนต่อจากนี้ที่จะสร้าง
สิ่งที่ transaction ต้องมี และ kaen-log ไม่มีสักอย่าง มีอยู่สามชิ้น
- transaction coordinator ตัวกลางที่จำสถานะของ transaction ที่เปิดค้างอยู่
- topic ภายในชื่อ
__transaction_stateที่Topic.javaประกาศไว้คู่กับ__consumer_offsets - control record ใน partition ของผู้ใช้ ที่บอกว่า transaction นั้นจบแบบไหน
control record ไม่ใช่ของที่เก็บแยก มันเป็น record จริงที่นั่งอยู่ใน partition เดียวกับข้อมูล
ControlRecordType.java ของรุ่น 4.1.0 ประกาศค่าของมันไว้ว่า ABORT((short) 0) และ COMMIT((short) 1)
ค่าคู่นี้ต้องอ่านจาก code ไม่ใช่จาก KIP-98 เพราะร้อยแก้วของ KIP เขียนคู่นี้สลับกัน ตัวที่ต้อง compile คือตัวที่ถูก
ฝั่งผู้อ่านมี config หนึ่งตัวที่เปลี่ยนขอบของการมองเห็น คือ isolation.level
ค่าปริยายของมันคือ read_uncommitted และเอกสารของ ConsumerConfig เขียนว่าในโหมด read_committed
consumer.poll() จะคืนข้อความได้แค่ถึง last stable offset ซึ่ง “is the one less than the offset
of the first open transaction”
flowchart LR
subgraph L["log ของ orders-0 · เรียงตามลำดับ"]
direction LR
S1["record ที่ commit แล้ว"] --> S2["record ของ transaction ที่ยังเปิดค้าง"] --> S3["record ที่ยังไม่ถึงเส้น committed"]
end
LSO["LSO · read_committed อ่านได้ถึงตรงนี้"] -.-> S1
HW["HW · read_uncommitted อ่านได้ถึงตรงนี้"] -.-> S2
LEO["LEO · ตำแหน่งที่จะเขียนตัวถัดไป"] -.-> S3
N["kaen-log สร้าง LEO กับ HW · เอ่ยชื่อ LSO เฉยๆ"]
คำบรรยายภาพ: ภาพนี้เป็นตัวอย่าง ไม่ใช่ผลรัน · เส้นสามเส้นเรียงกันเสมอในทิศทางเดียว โดย LSO ไม่มีวันเลย HW และ HW ไม่มีวันเลย LEO · transaction ที่เปิดค้างหนึ่งตัวจึงตรึง LSO ไว้กับที่ และผู้อ่านโหมด read_committed ทุกรายบน partition นั้นหยุดตามไปด้วย
ผลรันของ demo ปิดท้ายด้วยการเอ่ยชื่อทั้งสามเส้น และประกาศว่าอันไหนสร้างจริง
== transaction · วาดไว้ ไม่ได้สร้าง ==การกันซ้ำไม่ใช่ exactly-once · transaction ต้องมี coordinatorมี topic ภายในชื่อ __transaction_state และมี control record ใน partition ของผู้ใช้ทั้งหมดนั้นตั้งอยู่บนการทำ replication ของบทที่ 8 ขึ้นไปอีกชั้นเส้นแบ่งสามเส้นเมื่อ transaction มีจริง: LEO >= HW >= LSO LEO offset ตัวถัดไปที่จะเขียน HW เส้น committed · read_uncommitted อ่านได้ถึงตรงนี้ LSO offset ของ transaction ที่เปิดค้างตัวแรก · read_committed อ่านได้ถึงตรงนี้kaen-log สร้างสองเส้นแรก และเอ่ยชื่อเส้นที่สามเฉยๆและ exactly-once ไม่เคยเป็นสมบัติของการส่ง · มันเป็นสมบัติของ **ผลลัพธ์**ที่ปลายทางซึ่งกันซ้ำเป็น · การตัดบัตรหรือส่ง SMS ไม่ได้อยู่ในความคุ้มครองนี้กลไกที่ Kafka ใช้ประกอบ exactly-once ขึ้นมาจริงๆ อธิบายได้ในประโยคเดียว คือมันเขียนตำแหน่งของผู้อ่านลง topic ภายใน แล้วมัดการเขียนนั้นไว้ใน transaction เดียวกับผลลัพธ์
เอกสารออกแบบเขียนไว้ตรงตัวว่า “we can write the offset to Kafka in the same transaction as the output topics receiving the processed data” และย้ำว่าในระบบนี้ผู้ที่ทำ transaction ได้คือ producer ไม่ใช่ consumer
นั่นแปลว่า exactly-once ของ Kafka เป็นเรื่องของการอ่านแล้วเขียนกลับเข้า Kafka มันไม่ได้ครอบผลที่เกิดกับระบบข้างนอกเลยแม้แต่กรณีเดียว
ของที่บทนี้ไม่ได้สร้าง
หัวข้อที่มีชื่อว่า “ของที่บทนี้ไม่ได้สร้าง”หัวข้อนี้ประกาศเส้นขอบเขตของ
src/idem.rsแล้วส่งต่อไปบทสุดท้าย
- ไม่มี transaction และไม่มี coordinator ทั้งหมดถูกวาดไว้ในหัวข้อก่อนหน้าเท่านั้น
- ไม่มีการกันตัวเก่าออกด้วย epoch kaen-log ใช้ epoch เป็นตัวประกอบของการเทียบ ไม่ได้ใช้กันคนเก่า
- ไม่มี snapshot ของสถานะ producer ลง disk สถานะสร้างขึ้นใหม่จากการกวาดท่อนสุดท้ายเท่านั้น
- ไม่มีการหมดอายุตามเวลา Kafka ล้างสถานะของ producer ที่เงียบไปนาน ส่วน kaen-log ไม่ล้าง
ข้อแรกเป็นข้อเดียวที่ต้องการชั้นที่บทนี้ยังไม่มี อีกสามข้อเป็นการตัดขอบเขตของบทนี้เอง transaction ตั้งอยู่บนการทำสำเนาซึ่งเป็นเนื้อของบทถัดไป
บทที่ 8 จะเอา log ก้อนเดิมนี้ไปทำสำเนาไว้อีกที่ แล้วนิยามเส้น committed ขึ้นมาจากสำเนาที่ตามทัน ตอนนั้นเส้น HW ในภาพข้างบนจะเลิกเดินตามปลาย log และกลายเป็นค่าที่คำนวณจากสำเนา
- Apache Kafka 4.1 — Design (ตรวจแล้ว 2026-09-03) — นิยามสามข้อของการส่ง ประโยคเตือนเรื่อง fine print ประโยคที่ระบุว่าตั้งแต่ 0.11.0.0 broker แจก id ให้ producer แล้วกันซ้ำด้วย sequence ประโยคเรื่องการเขียนตำแหน่งผู้อ่านใน transaction เดียวกับผลลัพธ์ และประโยคที่จำกัดขอบของ exactly-once ไว้ที่ระบบปลายทางซึ่งต้องร่วมมือด้วย
- KIP-98 — Exactly Once Delivery and Transactional Messaging (ตรวจแล้ว 2026-09-03) — ประโยคที่ producer ทุกตัวได้ PID ตอน initialize และประโยคที่ sequence เริ่มจากศูนย์ โดยมีหนึ่งตัวนับต่อหนึ่ง topic partition
- ProducerStateEntry.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) —
NUM_BATCHES_TO_RETAIN = 5javadoc ที่อธิบายการดันตัวหน้าออกเมื่อคิวเต็ม และfindDuplicateBatchที่ตัดด้วย epoch ก่อนแล้วหาช่วง sequence ที่ตรงเป๊ะ - BatchMetadata.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — field สี่ตัว
lastSeqlastOffsetoffsetDeltatimestampกับfirstSeq()และfirstOffset()ที่คำนวณกลับจากoffsetDelta - UnifiedLog.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — สาขา
result.maybeDuplicateที่ตอบด้วย offset ชุดเดิม และข้ามสาขาที่เรียกlocalLog.append(...)ไปทั้งก้อน - ProducerAppendInfo.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — จุดที่โยน
OutOfOrderSequenceExceptionและตัวinSequenceที่นิยามความต่อเนื่องว่าnextSeq == lastSeq + 1L - ProducerStateManager.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — file
.snapshotที่ Kafka มีและ kaen-log ไม่มี กับisProducerExpiredที่ล้างสถานะของ producer ที่เงียบเกินproducerIdExpirationMsและเป็นหนึ่งในสาม file ฝั่ง broker ที่ไม่มีทางเดินไหนโยนรหัส 46 เลย - Errors.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — รหัส 45
OUT_OF_ORDER_SEQUENCE_NUMBERและรหัส 46DUPLICATE_SEQUENCE_NUMBERพร้อมข้อความของทั้งคู่ - Sender.java tag 4.1.0 (ตรวจแล้ว 2026-09-03) — ฝั่ง client ที่จัดการรหัส 46 เอง โดยตอบผู้ใช้ว่าสำเร็จ แล้วไม่คืน offset กับ timestamp ที่ถูกต้อง
- ProducerConfig.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — เพดาน
max.in.flight.requests.per.connectionที่ผูกกับจำนวน batch ที่ broker เก็บไว้ และประโยคที่ว่าถ้าไม่มีtransactional.idก็ได้แค่ idempotent delivery - ConsumerConfig.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — ค่าปริยาย
read_uncommittedและนิยามของ last stable offset ว่าอยู่ก่อน offset ของ transaction ที่เปิดค้างตัวแรก - ControlRecordType.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) —
ABORT((short) 0)และCOMMIT((short) 1)ซึ่งเป็นค่าที่ต้องเชื่อ แทนร้อยแก้วของ KIP-98 ที่เขียนคู่นี้สลับกัน - DefaultRecordBatch.java tag 4.1.0 (ตรวจแล้ว 2026-09-03) — javadoc หัวข้อ On Compaction ที่ระบุว่ารูปแบบ v2 รักษา offset กับ sequence ตัวแรกและตัวสุดท้ายไว้ตอนกวาด log เพื่อให้สร้างสถานะของ producer ขึ้นใหม่ได้ตอนอ่าน log กลับ
- Topic.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) —
TRANSACTION_STATE_TOPIC_NAMEที่ประกาศไว้คู่กับGROUP_METADATA_TOPIC_NAMEในรายชื่อ topic ภายใน - Apache Kafka 0.11.0.0 บนแหล่งเก็บของ ASF (ตรวจแล้ว 2026-09-03) — วันที่ 2017-06-28 ของทุกชิ้นในรุ่นที่การกันซ้ำเข้ามาเป็นครั้งแรก
เช็กความเข้าใจ — บทที่ 7
ข้อ 1 / 3broker ได้ batch ที่มี producer id เดิม และช่วง sequence ตรงกับ batch ที่มันเก็บไว้ในหน้าต่างพอดี มันตอบอะไรกลับไป