เขียน · fsync · แล้วค่อยตอบ — และเหตุผลที่ Kafka ไม่ทำอย่างนั้น
คำขอที่วิ่งเข้ามาหา broker เป็น byte เรียงต่อกันเฉยๆ ไม่มีขอบ ไม่มีชนิด และไม่มีชื่อติดมาด้วย
งานแรกของ broker จึงไม่ใช่การเขียนอะไรลง disk แต่เป็นการตัดสินว่า byte ก้อนนี้จบตรงไหน และใครควรรับไปทำต่อ
บทนี้เดินตามคำขอ produce หนึ่งอันตั้งแต่ byte แรกจนถึงคำตอบ แล้วไปหยุดที่คำถามข้อเดียว คือ broker ควรรออะไรก่อนจะบอกว่าเขียนแล้ว
ข้ออ้างเดียวที่ต้องได้กลับไปคือ broker แจก offset แล้วเขียนลง page cache ก่อนเสมอ
จากนั้นจึงค่อยตัดสินว่าจะรออะไรต่อ acks เปลี่ยนเวลาที่ตอบ ไม่เคยเปลี่ยนสิ่งที่เขียน
crate ของคอร์สอยู่ที่ scripts/log-broker/ และชื่อ binary คือ klog
ทั้งคอร์สเขียนด้วย Rust standard library ล้วน
บล็อก code กับบล็อกผลรันทุกอันในบทนี้ถูกคัดมาจาก scripts/log-broker/src/produce.rs
จาก scripts/log-broker/src/partition.rs และจาก scripts/log-broker/expected/03-produce.txt
ทีละ byte โดยมีเทสต์เทียบไว้ ถ้าแก้ในบทเรียนอย่างเดียวโดยไม่แก้ที่ต้นทาง เทสต์จะแดงทันที
ผลรันทุกบรรทัดในบทนี้มาจากคำสั่งเดียว คือ
klog produce-demo --dir data/orders-0 --acks 1 --batches 3 --now-ms 1700000000000 --truncate-tail
ซึ่งเทสต์รันซ้ำแล้วเทียบผลทีละ byte ทุกครั้ง
12 byte แรกส่งงานต่อได้ โดยยังไม่ต้องแตะ body
หัวข้อที่มีชื่อว่า “12 byte แรกส่งงานต่อได้ โดยยังไม่ต้องแตะ body”หัวข้อนี้กางส่วนหัวคงที่ของคำขอ แล้วบอกว่าทำไม broker ตัดสินใจได้ก่อนอ่านเนื้อ
คอร์ส kaen-kvstore สอนการหุ้ม frame ด้วยความยาวนำหน้าไว้แล้ว คอร์สนี้ถือว่ารู้ และเติมของที่ kvstore ไม่มี คือส่วนหัวที่บอกว่า frame นี้เป็นคำขอชนิดไหน
request headerrequest headerส่วนหัวคงที่ของทุกคำขอบนสาย บอกชนิดคำขอ รุ่นของ protocol และเลขกำกับที่ใช้จับคู่คำตอบกลับ ของ Kafka มีสี่ช่อง และสามช่องแรกกว้างคงที่ทั้งหมด ความกว้างคงที่ตรงนี้ไม่ใช่รสนิยม มันคือสิ่งที่ทำให้อ่านแค่ต้น frame แล้วส่งงานต่อได้เลย
เอกสาร protocol ของ Kafka เขียนกติกาของขอบไว้ตรงตัวว่า “The client can read requests by first reading this 4 byte size as an integer N, and then reading and parsing the subsequent N bytes of the request”
และเขียนเรื่องสองช่องถัดมาไว้ว่า “These two 16-bit numbers, when taken together, uniquely identify the schema of the message to follow”
#[derive(Clone, Debug, PartialEq, Eq)]pub struct RequestHeader { pub api_key: i16, pub api_version: i16, pub correlation_id: i32, pub client_id: Option<String>,}
/// `size i32` + `api_key i16` + `api_version i16` + `correlation_id i32` = 12/// อ่านแค่นี้ก็ส่งงานต่อได้ โดยยังไม่ต้องแตะ body สักไบต์pub const ROUTING_PREFIX_LEN: usize = 12;ช่องที่สี่คือ client_id ซึ่งเป็น NULLABLE_STRING ความยาวไม่คงที่
มันจึงอยู่ท้ายสุด และอยู่นอกช่วงที่ broker ต้องอ่านเพื่อจะรู้ว่าใครควรรับงานนี้ไป
klog พิมพ์ frame จริงออกมาให้ดูทั้งหัว
== frame หนึ่งอัน · 12 byte แรกส่งงานต่อได้โดยไม่ต้องแตะ body ==frame ทั้งอันยาว 190 byte (4 byte ของความยาว + 186 byte ของ payload)00000000 00 00 00 ba 00 00 00 0d 00 00 00 07 00 0d 6b 61 |..............ka|00000010 65 6e 2d 70 72 6f 64 75 |en-produ| 00..04 size 186 04..06 api_key 0 (0 = produce) 06..08 api_version 13 08..12 correlation_id 7 12.. client_id NULLABLE_STRING · ยาว i16 แล้วตามด้วย byte00 00 00 ba คือความยาวของ payload ส่วน 00 00 ถัดมาคือ api key ของ produce ซึ่งเป็นศูนย์
แล้ว 00 0d คือ api version และ 00 00 00 07 คือเลขกำกับที่ใช้จับคู่คำตอบ
เลขกำกับตัวนั้นมีชื่อว่า correlation_id และมันมีอยู่เพราะคำตอบวิ่งกลับมาบนสายเดียวกัน
client จึงต้องรู้ว่าคำตอบที่ได้เป็นของคำขออันไหน
เอกสารหน้าเดียวกันรับประกันอีกข้อ คือบนสาย TCP เส้นเดียว คำขอถูกประมวลผลตามลำดับที่ส่งเข้ามา และคำตอบวิ่งกลับตามลำดับนั้นด้วย ลำดับที่ว่านี้เองคือของที่ประโยคถัดมารักษาไว้ “The broker’s request processing allows only a single in-flight request per connection in order to guarantee this ordering”
header v2 มีช่องที่หน้าเอกสารไม่ได้วาดไว้
หัวข้อที่มีชื่อว่า “header v2 มีช่องที่หน้าเอกสารไม่ได้วาดไว้”หัวข้อนี้ชี้ที่ผิดหนึ่งจุดในเอกสารของ Kafka แล้วบอกว่าเชื่อ file ไหนแทน
หน้า protocol ที่ render แล้วของ Kafka วาด request header v2 ไว้สี่ช่องเท่ากับ v1 พอดี ถ้าเขียนตามหน้านั้น จะได้ header ที่สั้นกว่าที่ schema ของโครงการประกาศไว้หนึ่ง byte
ตัวที่ถูกอยู่ที่ RequestHeader.json ใน source tree ของโครงการเอง
มันประกาศ "validVersions": "1-2" และ "flexibleVersions": "2+"
flexibleVersions ที่ครอบ v2 แปลว่า v2 มี tagged field ต่อท้าย
แม้จะยังไม่มี tag สักตัวก็ต้องเขียนตัวนับที่เป็นศูนย์ลงไป และตัวนับนั้นกินที่หนึ่ง byte
pub fn encode(&self, header_version: i16) -> Vec<u8> { let mut out = Vec::new(); out.extend_from_slice(&self.api_key.to_be_bytes()); out.extend_from_slice(&self.api_version.to_be_bytes()); out.extend_from_slice(&self.correlation_id.to_be_bytes()); put_nullable_string(&mut out, self.client_id.as_deref()); if header_version >= HEADER_V2 { // ⚠ หน้าเอกสาร protocol ที่ render แล้วของ Kafka **แสดง header v2 ผิด** // มันวาดสี่ field เท่ากับ v1 และไม่มี TAG_BUFFER · แต่ RequestHeader.json // ประกาศ "flexibleVersions": "2+" ซึ่งแปลว่า v2 มี tagged field ต่อท้าย // · เขียนตามหน้า HTML แล้ว broker จริงจะปฏิเสธ frame นั้น · เชื่อ .json put_unsigned_varint(&mut out, 0); } out }== กับดักของ header v2 ==header v1 ยาว 23 byte · header v2 ยาว 24 byteต่างกัน 1 byte คือ TAG_BUFFER ที่ว่างเปล่า (unsigned varint ค่า 0)payload ทั้งอันจึงยาว 186 กับ 187 byte ตามลำดับหน้า protocol ที่ render แล้วของ Kafka วาด v2 ไว้เท่ากับ v1 และไม่มี TAG_BUFFERแต่ RequestHeader.json ประกาศ flexibleVersions 2+ · เชื่อ .json ไม่ใช่หน้า HTMLและ varint ของ protocol เป็นแบบ unsigned ส่วนของ record เป็น zigzag · คนละตัวdecode payload v2 กลับได้ · correlation_id 7 · acks 1ความต่างทั้งหมดคือหนึ่ง byte และเป็น byte ที่มองไม่เห็นในหน้าเอกสาร
ตัวนับ tag ตัวนั้นเป็น unsigned varint ซึ่งเป็นคนละตัวกับ zigzag varint ที่ บทที่ 1 กางไว้สำหรับช่องข้างใน record
เอาสองอย่างนี้ไปใช้ routine เดียวกันเมื่อไหร่ พังทั้งสองฝั่ง เพราะ zigzag เขียนหนึ่งเป็น 0x02
ส่วน unsigned varint เขียนหนึ่งเป็น 0x01
ของที่ตรวจได้มีสองชิ้น หน้า protocol ของรุ่น 4.3 วาด request header v2 ไว้สี่ช่องเท่ากับ v1
และ RequestHeader.json ที่ tag 4.1.0 ประกาศ "flexibleVersions": "2+" ไว้ตรงๆ
สองชิ้นนี้ไม่ใช่รุ่นเดียวกัน และคอร์สนี้ไม่ได้เปิด broker จริงมาตัดสิน สิ่งที่พูดได้จึงเป็นแค่ว่าเอกสารที่เผยแพร่กับ schema ที่อยู่ใน source tree บอกไม่ตรงกัน
กฎที่ใช้ได้ต่อคือเลือกชิ้นที่อยู่ใกล้ code ที่รันจริงที่สุด .json ตัวนั้นถูก ship ไปกับ
broker ส่วนหน้า HTML เป็นเอกสารของรุ่นที่เผยแพร่แล้ว
และเมื่อเลือกผิด สิ่งที่ได้ไม่ใช่คำเตือน แต่เป็น frame ที่ผิดรูปตั้งแต่ byte สุดท้ายของ header ซึ่งไม่มีตัวไหนบอกตอนคอมไพล์
ความยาวที่อีกฝั่งบอกมา ต้องตรวจก่อนจอง memory
หัวข้อที่มีชื่อว่า “ความยาวที่อีกฝั่งบอกมา ต้องตรวจก่อนจอง memory”หัวข้อนี้ผูกกฎสองข้อของ
std::ioเข้ากับการอ่าน frame หนึ่งอัน
ความยาวที่นำหน้า frame เป็นเลขที่อีกฝั่งเป็นคนบอก ไม่ใช่เลขที่เรานับเอง
buf.resize(n, 0) ที่รับ n มาตรงๆ จึงเป็นการให้ใครก็ได้บนเครือข่ายสั่งจองหน่วยความจำเท่าไรก็ได้
กติกาคือตรวจให้จบก่อน แล้วค่อยจอง
/// อ่านหนึ่ง frame · **ตรวจความยาวก่อนจอง memory** เพราะเลขนั้นมาจากอีกฝั่ง////// `read_exact` ไม่ใช่ `read` · `read` คืน byte น้อยกว่าที่ขอได้ตามกฎหมาย/// แม้จะยังไม่จบ stream · การอ่าน frame ด้วย `read` คือ bug ที่ผ่านเทสต์บนเครื่องเราpub fn read_frame<R: Read>(r: &mut R, buf: &mut Vec<u8>) -> io::Result<()> { let mut len_bytes = [0u8; 4]; r.read_exact(&mut len_bytes)?; let n = i32::from_be_bytes(len_bytes); let n = usize::try_from(n).map_err(|_| bad(format!("ความยาว frame ติดลบ: {n}")))?; if n > MAX_FRAME { return Err(bad(format!("frame ยาว {n} byte เกินเพดาน {MAX_FRAME}"))); } buf.clear(); buf.resize(n, 0); r.read_exact(buf)}เพดาน MAX_FRAME ของ kaen-log สะท้อน socket.request.max.bytes ของ Kafka
ซึ่งเอกสาร broker config บอกว่าตั้งต้นที่ 104857600 byte และอธิบายตัวเองไว้แค่ว่า
เป็นจำนวน byte มากที่สุดในหนึ่งคำขอ
kaen-log ยืมตัวเลขนั้นมาเพื่อให้ขอบของมันไม่หลวมกว่าของจริง ส่วนคำถามว่า broker จริง ทำอะไรกับสายที่ส่งคำขอเกินเพดานมา เป็นของที่คอร์สนี้ไม่ได้ตรวจ
read_exact ในบรรทัดถัดมาก็ไม่ใช่รสนิยมเหมือนกัน เอกสารของ Read::read เขียนไว้ว่า
“It is not an error if the returned value n is smaller than the buffer size,
even when the reader is not at the end of the stream yet”
แปลว่า read คืน byte ไม่ครบได้ทั้งที่ stream ยังไม่จบ
การอ่าน frame ด้วย read จึงเป็น bug ที่ผ่านทุกเทสต์บนเครื่องเรา แล้วไปพังบนเครือข่ายจริง
== ตรวจความยาวก่อนจอง memory ==frame ที่อ้างว่ายาว 104857601 byte ถูกปฏิเสธ · InvalidDataframe ที่บอกว่ายาว 8 byte แต่มีจริง 3 · UnexpectedEofความยาวเป็นเลขที่อีกฝั่งเป็นคนบอก · ตรวจก่อน resize ไม่ใช่หลังสองบรรทัดนั้นเป็นความผิดพลาดคนละแบบ และ ErrorKind ที่ได้ก็คนละตัว
- frame ที่อ้างความยาวเกินเพดาน ถูกปฏิเสธด้วย
InvalidDataตั้งแต่ก่อนจอง memory - frame ที่บอกว่ายาว 8 byte แต่มีจริง 3 ถูกปฏิเสธด้วย
UnexpectedEofตอนอ่านเนื้อ
UnexpectedEof ที่เกิดกลาง frame เป็นเรื่องจบสำหรับสายเส้นนั้น
เพราะเอกสารของ read_exact เขียนไว้ว่า “If this function returns an error,
it is unspecified how many bytes it has read”
เมื่อไม่รู้ว่าอ่านไปกี่ byte ก็ไม่มีทางรู้ว่าขอบของ frame ถัดไปอยู่ตรงไหน ทางเดียวที่เหลือคือทิ้งสายนั้น ไม่ใช่ลองอ่านต่อ
batching เป็นการตัดสินใจเรื่องรูปแบบ ไม่ใช่การปรับจูน
หัวข้อที่มีชื่อว่า “batching เป็นการตัดสินใจเรื่องรูปแบบ ไม่ใช่การปรับจูน”หัวข้อนี้บอกว่าทำไมการรวม record เป็นชุดถึงอยู่ในรูป byte ไม่ได้อยู่ที่ค่า config
batchingbatchingการรวมหลาย record เป็น batch เดียวก่อนส่ง · ซื้อ throughput มาโดยจ่ายเป็น latency ของ record ตัวแรกในชุด มักถูกเล่าเป็นการปรับจูน คือถ้าอยากได้ throughput ก็เพิ่มขนาดชุด แต่ในรูป byte ของ Kafka มันไม่ใช่ตัวเลือก มันคือหน่วยที่ทั้งรูปแบบถูกออกแบบมารอบมัน
หนึ่ง batch มีหัวหนึ่งหัวและ checksum หนึ่งตัว ส่วน record ข้างในเก็บแต่ส่วนต่างจากหัว ทั้ง offset และ timestamp
ผลคือ record ตัวที่สองในชุดไม่ได้เก็บ offset ของตัวเอง มันเก็บเลขหนึ่ง และเลขหนึ่งในรูป zigzag varint กินที่หนึ่ง byte
record ที่อยู่ตัวเดียวในชุดจึงแพงกว่า record ที่อยู่รวมกับเพื่อน ไม่ใช่เพราะ syscall แต่เพราะมันต้องแบกหัวทั้งหัวคนเดียว
ฝั่ง producer ของ Kafka มีสองค่าที่แข่งกันเป็นตัวปิดชุด และเอกสาร producer config เขียนความสัมพันธ์ไว้ตรงตัว
batch.sizeตั้งต้นที่ 16384 และเป็นเพดานของชุดlinger.msตั้งต้นที่ 5 เป็นเวลารอที่ยอมจ่ายเพื่อให้ชุดโตขึ้น และเอกสารบอกว่าค่านี้เปลี่ยนจากศูนย์มาเป็นห้าใน Kafka 4.0- ชุดที่เต็มก่อนชนะเวลารอ ประโยคนี้อยู่ในคำอธิบายของ
linger.msเอง คำว่า this setting ในนั้นจึงหมายถึงเวลารอ ไม่ใช่ขนาดชุด “once we getbatch.sizeworth of records for a partition it will be sent immediately regardless of this setting” batch.sizeที่เป็นศูนย์ ปิด batching ทิ้งทั้งหมด ตามที่เอกสารเขียนว่า “a batch size of zero will disable batching entirely”
ราคาของ batching อยู่ที่ record ตัวแรกของชุดเสมอ มันถูกกักไว้รอเพื่อนที่ยังมาไม่ถึง สิ่งที่ซื้อมาได้คือ throughputthroughputจำนวน operation ต่อวินาทีที่ระบบรับหรือจ่ายได้ · ใน broker มันคือของที่ batching ซื้อมา และเป็นอีกฝั่งของ latency ของทั้งกลุ่ม สิ่งที่จ่ายไปคือ latency ของตัวแรก
เอกสารของ linger.ms เทียบตัวเองไว้ว่า “This can be thought of as analogous to
Nagle’s algorithm in TCP” ซึ่งเป็นการเตือนที่มีประโยชน์ เพราะ Nagle ยังทำงานอยู่ข้างล่างอีกชั้น
การหน่วงสองชั้นซ้อนกันเป็นของที่มีอยู่จริง และตัวที่อยู่ข้างล่างไม่ได้อ่านค่าที่เราตั้งไว้ข้างบน
broker เป็นคนแจก offset ไม่ใช่ producer
หัวข้อที่มีชื่อว่า “broker เป็นคนแจก offset ไม่ใช่ producer”หัวข้อนี้แสดงว่าการแจกลำดับเกิดขึ้นที่ไหน และหนึ่งคำตอบครอบคลุมกี่ record
producer ประกอบ batch ขึ้นมาโดยที่ยังไม่รู้ว่า record ของมันจะไปนั่งลำดับไหน บทที่ 1 กางไว้แล้วว่าการประทับลำดับเป็นการเขียนทับ 8 byte ที่หัว batch โดยไม่ต้องคิด checksum ใหม่
บทนี้เอาการประทับนั้นมาวางไว้บนทางเดินจริง คือ append ตัวเดียวของทั้ง crate
/// ประทับ base offset ลงบน **สำเนา** ของ `batch.raw` แล้วเขียนสำเนานั้น /// คืน base offset ตัวแรกที่แจกไปในรอบนี้ · **broker เป็นคนแจก offset ไม่ใช่ producer** /// /// `now_ms` คือนาฬิกาของ broker ตอน append และใช้กับเงื่อนไข roll ตามเวลา **เท่านั้น** /// · timestamp ของ record เป็น CreateTime มาจาก producer จึงไม่ถูกแตะที่นี่ pub fn assign_and_append(&mut self, batches: &[Batch], now_ms: i64) -> io::Result<u64> { // จำนวน batch เป็นเลขที่อีกฝั่งเป็นคนบอก · `count = 0` ผ่าน `decode` มาได้ // ถ้าปล่อยผ่าน ทางนี้จะคืน "สำเร็จ" โดยไม่ได้แจก offset ให้อะไรเลย // แล้ว `acks=-1` ที่คิด `log_end - 1` ต่อจากนั้นจะลบ u64 ทะลุศูนย์ if batches.is_empty() { return Err(bad("คำขอที่ไม่มี batch เลยสักก้อน")); } let assigned_base = self.next_offset; for batch in batches { let count = u64::try_from(batch.header.records_count).map_err(|_| bad("records_count ติดลบ"))?; if count == 0 { return Err(bad("batch ที่ไม่มี record เลย")); } let last = self.next_offset + count - 1;
// Kafka เขียนทับ 8 byte ใน buffer ที่รับมา แล้วไม่คิด CRC ใหม่ // Rust ให้ slice มาแบบอ่านอย่างเดียว จึงสำเนาก่อนแล้วประทับบนสำเนา let mut bytes = batch.raw.clone(); BatchHeader::stamp_last_offset(&mut bytes, last, batch.header.last_offset_delta)?; BatchHeader::stamp_leader_epoch(&mut bytes, self.leader_epoch);
self.roll_if_needed(len_to_u64(bytes.len()), last, now_ms)?;
// ตรวจ **ก่อน** บวกตัวนับ · เหมือน LogSegment.append ทุกตัวอักษร let add_index_entry = self.bytes_since_last_index_entry > self.config.index_interval_bytes; let seg = self.segments.last_mut().expect("log มีอย่างน้อยหนึ่งท่อนเสมอ"); seg.append(&bytes, last, batch.header.max_timestamp, add_index_entry)?; if add_index_entry { self.bytes_since_last_index_entry = 0; } self.bytes_since_last_index_entry += len_to_u64(bytes.len()); self.next_offset = last + 1; // มี replica ตัวเดียว ISR จึงมีแต่ leader เอง และเส้น committed // เดินตาม log end ทุกก้าว · บทที่ 8 ปิดข้อนี้เมื่อมี replica ตัวที่สอง if self.hw_follows_log_end { self.high_watermark = self.next_offset; } } Ok(assigned_base) }มีข้อหนึ่งที่ Rust บังคับให้ต่างจาก Kafka · ฝั่ง Kafka เขียนทับ 8 byte ลงบน buffer
ที่รับมาได้เลย ส่วน batch.raw ที่นี่ยืมมาแบบอ่านอย่างเดียว จึงต้องสำเนาก่อนแล้วประทับบนสำเนา
การสำเนาเป็นราคาที่จ่ายจริง และเป็นราคาที่ตรงไปตรงมากว่าการหาทางเขียนทับของที่ยืมมา
now_ms ที่รับเข้ามาเป็นนาฬิกาของ broker และมันมีผลกับเงื่อนไข roll ตามเวลาเท่านั้น
timestamp ของ record เป็น CreateTime ที่มาจาก producer จึงไม่ถูกแตะที่นี่เลย
demo ยิงสาม batch เข้าไป ชุดละไม่เท่ากัน แล้วพิมพ์สิ่งที่ broker ตอบกลับมา
== broker เป็นคนแจก offset · หนึ่งคำตอบต่อหนึ่ง batch ==acks ที่ใช้ในหัวข้อนี้ = 1 batch record base_offset ที่ได้ log end หลังเขียน error 0 2 0 2 0 1 3 2 5 0 2 4 5 9 0log end offset สุดท้าย 9base_offset เป็นของ batch ไม่ใช่ของ record · record ที่เหลือหาเอาจาก offset_deltaชุดแรกมีสอง record ได้ base_offset 0 แล้ว log end offset ขยับไปที่ 2 ชุดที่สองมีสาม record ได้ base_offset 2 และ log end offset ไปที่ 5
ตัวเลขในตารางนี้เป็นเลขของชุด ไม่ใช่เลขของ record คำตอบหนึ่งอันบอก base_offset มาตัวเดียว ส่วน record ที่เหลือในชุดเดียวกัน client หาเอาเองจาก offset_delta ที่อยู่ข้างใน
เอกสาร protocol ของ Kafka วาง base_offset ไว้เป็นช่องระดับ partition ในคำตอบเหมือนกัน
ไม่ได้มีช่องต่อ record ให้เลยสักช่อง
acks เปลี่ยนเวลาที่ตอบ ไม่เคยเปลี่ยนสิ่งที่เขียน
หัวข้อที่มีชื่อว่า “acks เปลี่ยนเวลาที่ตอบ ไม่เคยเปลี่ยนสิ่งที่เขียน”หัวข้อนี้ไล่ค่าของ acks ทีละค่า แล้วชี้ว่าลำดับการทำงานไม่ขยับตามมันเลย
acksacksค่าที่ producer สั่งว่า broker ต้องตอบกลับเมื่อไหร่ · มันเปลี่ยนแค่จังหวะการตอบ ไม่ได้เปลี่ยนวิธีเขียนลง log รับได้สามค่าเท่านั้นบน wire และเลขสามตัวนั้นไม่ได้แปลว่า “เอากี่สำเนา”
/// สาม value เดียวที่ `acks` รับได้บน wire · ไม่มี `acks=2`/// "จะเอากี่ replica" เป็นหน้าที่ของ `min.insync.replicas` ไม่ใช่ของ `acks`pub const ACKS_NONE: i16 = 0;pub const ACKS_LEADER: i16 = 1;pub const ACKS_ALL: i16 = -1;
/// **ตาราง error ของ kaen-log เป็นของตัวเอง** รอบวิจัยของคอร์สนี้ไม่ได้ยืนยันเลขจริง/// ในตารางของ Kafka จึงไม่มีบทไหนอ้างว่าเลขพวกนี้เป็นเลขบน wire ของ Kafkapub const ERR_NONE: i16 = 0;pub const ERR_INVALID_REQUIRED_ACKS: i16 = 1;pub const ERR_NOT_ENOUGH_REPLICAS: i16 = 2;ERR_NONE ERR_INVALID_REQUIRED_ACKS และ ERR_NOT_ENOUGH_REPLICAS เป็นเลขที่ crate นี้
ตั้งขึ้นเอง เพื่อให้ demo พิมพ์อะไรออกมาได้
รอบวิจัยของคอร์สนี้ไม่ได้ตามหาตาราง error code จริงของ Kafka มาเทียบ จึงไม่มีบทไหนอ้างว่าเลขพวกนี้เป็นเลขบน wire ของ Kafka ถ้าต้องใช้เลขจริง ให้ไปเปิดของจริง ไม่ใช่เดาจากที่นี่
ฝั่ง Kafka ตรวจสามค่านี้ด้วย function บรรทัดเดียวใน ReplicaManager.scala คือ
requiredAcks == -1 || requiredAcks == 1 || requiredAcks == 0
และคอมเมนต์ใน function ที่ตอบ error ตัวนั้นเขียนไว้ว่าถ้าค่าหลุดออกนอกช่วงนี้ แปลว่า client มีอะไรผิด
ให้ตอบ error แล้วไม่ต้องรับคำขอนั้นเลย
ไม่มี acks=2 คำถามว่า “ต้องมีกี่สำเนาถึงจะนับว่า commit” ไม่ได้ถามที่ acks
มันถูกถามที่ min.insync.replicas ซึ่งเป็นค่าฝั่ง broker และเป็นเรื่องของบทที่ 8
/// `Ok(None)` คือ `acks=0` · **ไม่ส่ง frame ตอบกลับเลยสักไบต์**/// `io::Result` เป็นชั้นนอก เพราะการเขียนลง disk ล้มเหลวได้ก่อนจะถึงเรื่องว่าจะตอบอะไรpub fn handle_produce( p: &Partition, gate: &dyn ReplicaGate, req: &ProduceRequest, now_ms: i64,) -> io::Result<Option<ProduceResponse>> { if !matches!(req.acks, ACKS_NONE | ACKS_LEADER | ACKS_ALL) { let response = ProduceResponse::failed(req.header.correlation_id, ERR_INVALID_REQUIRED_ACKS); return Ok(Some(response)); }
// เขียนใต้ lock · **fsync อยู่นอก lock** เพราะ fsync กินเวลาระดับมิลลิวินาที // และการถือ lock คร่อมมันจะกั้น producer ทุกรายกับ fetch ที่จอดรออยู่ทุกตัว let (base, log_end) = { let mut log = p.log.lock().expect("log ถูก poison"); let base = log.assign_and_append(&req.batches, now_ms)?; (base, log.log_end_offset()) }; p.sync_all()?; p.publish_log_end(log_end);
let response = ProduceResponse { correlation_id: req.header.correlation_id, error_code: ERR_NONE, base_offset: crate::record::wire_i64(base), log_append_time_ms: -1, log_start_offset: 0, throttle_time_ms: 0, };
match req.acks { // ไม่มีคำตอบ · Kafka บอกเองว่าเมื่อเกิด error มันจะ **ปิด socket** แทนที่จะตอบ // เพราะ producer ไม่ได้รออะไรอยู่ · offset ที่ client เห็นจะเป็น -1 เสมอ ACKS_NONE => Ok(None), ACKS_LEADER => Ok(Some(response)), ACKS_ALL => Ok(Some(await_commit(gate, log_end - 1, req, now_ms).0)), other => Err(bad(format!("acks {other} ผ่านการตรวจมาได้อย่างไร"))), }}อ่านลำดับในนั้นแล้วจะเห็นว่าทั้งการเขียนและ sync_all จบไปก่อนจะถึง match req.acks
ตัว match ไม่ได้เลือกว่าจะเขียนอะไร มันเลือกว่าจะพูดอะไรกลับไปหลังเขียนเสร็จแล้ว
ค่าที่ไม่ถูกต้องถูกคัดออกก่อนหน้านั้นอีกชั้นหนึ่ง คำขอที่ไปถึง match ได้จึงเขียนลง log
ไปแล้วทุกอัน ไม่มีกิ่งไหนของ match ที่ย้อนกลับไปเขียนหรือไม่เขียน
sequenceDiagram
autonumber
participant P as producer
participant B as broker
participant D as disk
P->>B: Produce · acks เป็น 0 หรือ 1 หรือ -1
B->>B: แจกลำดับแล้วประทับลงหัว batch
B->>D: write_all · ไปถึงแค่ page cache
B->>D: sync_all · ทำนอก lock ของ log
alt acks เป็น 0
Note over B,P: ไม่ส่ง frame ตอบกลับเลยสักไบต์ · ถ้าพลาด Kafka ปิดสายทิ้ง
else acks เป็น 1
B->>P: ตอบทันทีพร้อม base_offset ของชุด
else acks เป็น -1
B->>B: ถาม ReplicaGate ว่าถึงเส้นที่ต้องการหรือยัง
B->>P: ยังไม่พร้อม จนกว่าบทที่ 8 จะเอา replica ตัวจริงมาใส่
end
คำบรรยายภาพ: ภาพนี้เป็นทางเดินของคำขอหนึ่งอัน ไม่ใช่ผลรัน · สามกิ่งล่างต่างกันแค่สิ่งที่ broker พูดกลับไป ส่วนสองขั้นบนคือการแจกลำดับกับการเขียน เกิดขึ้นเหมือนกันหมดทุกกิ่ง
demo ยิงคำขอสี่อันที่ต่างกันแค่ค่า acks แล้ววัด log end ก่อนกับหลังทุกครั้ง
== acks เปลี่ยนเวลาที่ตอบ ไม่เคยเปลี่ยนสิ่งที่เขียน == acks คำตอบ log end ก่อน → หลัง 0 ไม่ส่ง frame ตอบกลับเลย 9 → 10 1 ok · base_offset 10 10 → 11 -1 not_enough_replicas 11 → 12 2 invalid_required_acks 12 → 12acks=0 เขียนลง disk เหมือนกันทุกอย่าง แค่ไม่มีใครได้ยินคำตอบacks=2 ไม่ใช่ค่าที่รับได้ · log end ไม่ขยับ เพราะคำขอถูกปฏิเสธก่อนถึง diskacks=-1 ยังตอบ not_enough_replicas เพราะ gate ของบทที่ 1 ถึง 7 ตอบ"ยังไม่พร้อม" เสมอ · บทที่ 8 เป็นคนเอา replica จริงมาใส่สามแถวแรกขยับ log end ไปข้างหน้าหนึ่งขั้นเท่ากันหมด ทั้งที่คำตอบต่างกันคนละอย่าง
acks=0ไม่มี frame ตอบกลับเลย แต่ record ลงไปนั่งใน log เรียบร้อยacks=1ตอบพร้อม base_offset ของชุดที่เพิ่งเขียนacks=-1ตอบว่าnot_enough_replicasเพราะยังไม่มี replica ตัวที่สองให้รอ
แถวสุดท้ายคือ acks=2 ซึ่งไม่ใช่ค่าที่รับได้ และเป็นแถวเดียวที่ log end ไม่ขยับ
เพราะคำขอถูกปฏิเสธตั้งแต่ก่อนถึง disk
acks=0 เป็นแถวที่อ่านผิดง่ายที่สุด มันไม่ได้แปลว่าไม่เขียน มันแปลว่าไม่มีใครได้ยินคำตอบ
เอกสาร producer ของ Kafka บอกด้วยว่า offset ที่ client เห็นในกรณีนี้จะเป็นลบหนึ่งเสมอ
/// **ไม่มี function ชื่อ `park_until_hw` ใน crate นี้** · `acks=-1` เรียกผ่าน trait นี้/// และบทที่ 1 ถึง 7 ให้ตัว stub ที่ตอบ "ยังไม่พร้อม" เสมอpub trait ReplicaGate: Send + Sync { fn check_enough_replicas_reached(&self, required_offset: u64) -> bool;}
/// stub ของบทที่ 1 ถึง 7 · ยังไม่มี replica สักตัว จึงยังไม่มีอะไรถึง required_offset ได้pub struct NoReplicas;
impl ReplicaGate for NoReplicas { fn check_enough_replicas_reached(&self, _required_offset: u64) -> bool { false }}acks=all ถูกนิยามไว้กับเซตของ replica ที่ตามทัน ไม่ใช่กับ disk ของเครื่องเดียว
kaen-log ในบทนี้มี replica อยู่ตัวเดียวคือตัวมันเอง จึงไม่มีอะไรที่ acks=-1 จะรอได้อย่างซื่อสัตย์
ทางที่ง่ายกว่าคือให้ acks=-1 ตอบ ok เหมือน acks=1 แล้วเดินหน้าต่อ
ซึ่งจะเป็นคำโกหกที่ผู้อ่านตรวจไม่ได้ เพราะผลรันจะสวยเท่ากันทั้งที่ไม่มีสำเนาอยู่จริง
crate จึงเลือกให้มันตอบ not_enough_replicas ทุกครั้ง จนกว่าบทที่ 8 จะเอา follower ตัวจริง
มาเสียบในช่องเดียวกันนี้ ช่องว่างที่ประกาศตัวเองว่าว่าง ปลอดภัยกว่าช่องที่แกล้งเต็ม
fsync ที่อยู่นอก lock
หัวข้อที่มีชื่อว่า “fsync ที่อยู่นอก lock”หัวข้อนี้แยกสิ่งที่ต้องทำใต้ lock ออกจากสิ่งที่ต้องทำนอก lock
fsyncfsyncบังคับ byte จาก page cache ลง storage จริง (`File::sync_all`/`sync_data`) · kaen-log เรียกมันเพราะยังไม่มี replica ส่วน Kafka จริงไม่เรียกบนทาง write คือการบังคับให้ byte ที่อยู่ใน page cache ลง storage จริง
คอร์ส kaen-kvstore สอนไว้แล้วว่า write_all ไปถึงแค่ page cache
และคอร์สนี้ถือว่ารู้
สิ่งที่คอร์สนี้เพิ่มคือคำถามว่าใครถือ lock อยู่ตอนที่ fsync ทำงาน
ถ้าเขียนตามสัญชาตญาณจะได้ p.log.lock().unwrap().sync_all() ซึ่งอ่านแล้วดูสะอาด
แต่บรรทัดนั้นถือ lock ของ log ไว้ตลอดการรอ storage
ผลคือ producer รายอื่นทุกรายและคำขอ fetch ที่จอดรออยู่ทุกอัน ต้องรอ fsync ของคนอื่นไปด้วย
/// ถือ lock ของ log แค่ตอน `dup` descriptor แล้ว fsync **นอก lock ทุกตัว** pub fn sync_all(&self) -> io::Result<()> { let file = { let log = self.log.lock().expect("log ถูก poison"); log.active_file()? }; file.sync_all() }active_file() ในบรรทัดกลางคือการเรียก File::try_clone หนึ่งครั้ง และ try_clone
ไม่ได้สำเนาเนื้อ file เอกสารของ std เขียนไว้ว่ามันคืน File ตัวใหม่ที่
“shares the same underlying file handle as the existing File instance”
การเรียก sync_all() บนตัวสำเนาจึงบังคับ byte ก้อนเดียวกันลง storage ไม่ใช่ของคนละก้อน
lock ถูกถือไว้แค่ช่วงที่สร้างสำเนานั้นขึ้นมา ส่วนการรอ storage เกิดขึ้นตอนที่ไม่มีใครถูกกั้น
เอกสารของ File::sync_all ยังเตือนอีกข้อที่คุ้มค่าจะจำ ว่าการปล่อย File ทิ้งเฉยๆ
“will ignore all errors” การไม่เรียก sync_all จึงไม่ใช่แค่ไม่รอ แต่เป็นการไม่รู้ด้วยว่าพลาดหรือเปล่า
durabilitydurabilityการรับประกันว่าข้อมูลที่ ack แล้วรอด crash (ต้องผ่าน fsync ก่อน) · ใน broker ข้อรับประกันนี้ย้ายจาก disk ไปอยู่กับจำนวนสำเนาได้ ของ kaen-log จึงมีที่อยู่ที่เดียว คือ disk ของเครื่องนี้ และมันต้องยืนยันได้ตอนที่คำตอบถูกส่งออกไป ไม่ใช่ตอนที่ process ตายไปแล้ว
คอมเมนต์ใน handle_produce เขียนว่า fsync กินเวลาระดับมิลลิวินาที ซึ่งเป็นเหตุผลของการออกแบบ
ที่ผู้เขียน crate ใส่ไว้ ไม่ใช่ผลวัดของคอร์สนี้
expected/03-produce.txt ไม่มีตัวเลขเวลาสักตัว เพราะ demo ทุกตัวของคอร์สนี้ต้องให้ผลเหมือนเดิม
ทุกรอบ และเวลาที่วัดจากเครื่องจริงไม่มีทางเหมือนเดิม
ถ้าอยากได้ตัวเลขของเครื่องตัวเอง ให้เขียน probe ที่วัดแล้วบันทึกผลไว้ คอร์ส kaen-kvstore บทที่ 7 ทำแบบนั้นไว้ทั้งบท
Kafka ไม่ fsync บนทาง write และนั่นคือการตั้งใจ
หัวข้อที่มีชื่อว่า “Kafka ไม่ fsync บนทาง write และนั่นคือการตั้งใจ”หัวข้อนี้กลับข้างสิ่งที่เพิ่งสร้าง แล้วบอกว่าทำไมของจริงถึงเลือกทางตรงข้าม
kaen-log fsync ก่อนตอบ ส่วน Kafka ไม่ทำอย่างนั้น และไม่ใช่เพราะลืม
ค่าตั้งต้นของ log.flush.interval.messages คือ 9223372036854775807
ซึ่งเท่ากับค่ามากที่สุดของ i64 ส่วน log.flush.interval.ms ไม่ได้ตั้งค่าไว้เลย
ทางเดิน append ของ Kafka มีบรรทัดที่เรียก flush อยู่จริง คือ
if (localLog.unflushedMessages() >= config().flushInterval) flush(false);
แต่ที่ค่าตั้งต้น เงื่อนไขนั้นเป็นจริงไม่ได้
นี่คือกิ่งที่มีอยู่ในโปรแกรม แต่เดินไปถึงไม่ได้ด้วยค่าตั้งต้นที่เอกสารเองแนะนำให้ใช้
เอกสารฝั่งปฏิบัติการของโครงการพูดตรงกว่านั้นอีก ว่า “We recommend using the default flush settings which disable application fsync entirely” และ “We generally feel that the guarantees provided by replication are stronger than sync to local disk”
เหตุผลที่โครงการให้ไว้ในเอกสารออกแบบคือ “we do not want to require the use of fsync on every write for our consistency guarantees as this can reduce performance by two to three orders of magnitude”
“two to three orders of magnitude” อ่านแล้วเหมือนตัวเลข แต่หน้านั้นไม่ได้แนบ benchmark ไม่ได้บอกเครื่อง ไม่ได้บอก filesystem และไม่ได้ลงวันที่ไว้เลย
มันจึงเป็นข้ออ้างที่โครงการใช้อธิบายทางเลือกของตัวเอง ไม่ใช่ผลการวัดที่ใครทำซ้ำได้ อ้างมันได้ในฐานะเหตุผลที่ Kafka ให้ไว้ ห้ามอ้างในฐานะตัวเลข
คอร์สนี้ไม่ได้จับเวลา fsync สักครั้ง จึงไม่มีบทไหนพูดว่าเร็วขึ้นหรือช้าลงเท่าไร
Kafka ย้าย fsync ออกจากทางที่ producer รออยู่จริงๆ ตอน roll segment คอมเมนต์ใน
UnifiedLog.java เขียนไว้ว่า “We avoid potentially-costly fsync call, since we acquire
UnifiedLog#lock here which could block subsequent produces in the meantime”
กฎที่อยู่ใต้คอมเมนต์นั้นเป็นกฎเดียวกับที่ Partition::sync_all ของ kaen-log ใช้
คืออย่าถือ lock คร่อมการรอ storage
สองประโยคที่ต้องอยู่ด้วยกัน kaen-log fsync ก่อนตอบเพราะมันมีเครื่องเดียวและยังไม่มีสำเนาที่อื่น disk จึงเป็นที่เดียวที่คำสัญญาของมันไปอยู่ได้
และ Kafka ไม่ fsync เพราะมันมีสำเนาอยู่หลายเครื่อง ความทนทานของมันจึงย้ายจาก disk ไปอยู่กับจำนวนสำเนาแทน ตัวเลือกที่ถูกของของเล่น เป็นคำอธิบายที่ผิดของ Kafka
WALWALwrite-ahead log: บันทึกลง log ให้ durable ก่อนถือว่า commit · log ของ broker ไม่ใช่ WAL เพราะมันไม่ได้วางอยู่หน้าโครงสร้างข้อมูลอื่น มันคือโครงสร้างนั้นเอง คือการเขียนลง log ให้ durable ก่อน แล้วค่อยไปแก้โครงสร้างข้อมูลจริงตามหลัง ฐานข้อมูลใช้รูปนี้เพราะ log ของมันเป็นทางผ่าน ปลายทางคือ B-tree หรือตารางที่อยู่อีกที่หนึ่ง
log ของ broker ไม่มีปลายทางแบบนั้น มันไม่ได้เขียนล่วงหน้าก่อนอะไร เพราะมันคือตัวข้อมูลเอง คำว่า “ahead” ในชื่อจึงไม่มีของให้ไปอยู่ข้างหน้า
ความต่างนี้มีผลจริงในบทถัดๆ ไป log ของฐานข้อมูลตัดทิ้งได้เมื่อ checkpoint ผ่านแล้ว ส่วน log ของ broker ตัดทิ้งได้ก็ต่อเมื่อ retention บอกให้ตัด ซึ่งเป็นเรื่องของบทที่ 6
ท้าย file ที่ขาด ต้องตัดทิ้งถึง batch ที่ครบตัวสุดท้าย
หัวข้อที่มีชื่อว่า “ท้าย file ที่ขาด ต้องตัดทิ้งถึง batch ที่ครบตัวสุดท้าย”หัวข้อนี้เปิด log ที่มีขยะค้างท้าย แล้วแสดงว่าการอ่าน offset ให้ถูกยังไม่พอ
เครื่องดับกลางการเขียนไม่ได้ทิ้ง log ที่เสียหายทั้งก้อนไว้ มันทิ้ง log ที่ดีทั้งก้อน บวกเศษของ batch สุดท้ายที่เขียนไม่จบ
truncate-tail recoverytruncate-tail recoveryการกู้คืนหลัง crash ด้วยการตัดหางที่เขียนไม่ครบทิ้ง แล้วเขียนต่อจากขอบของ record สมบูรณ์ตัวสุดท้าย คือการตัดเศษนั้นทิ้ง แล้วเขียนต่อจากขอบของ batch ที่ครบตัวสุดท้าย
Kafka เขียนกฎการตรวจไว้ในเอกสารว่า “A message entry is valid if the sum of its size and offset are less than the length of the file AND the CRC32 of the message payload matches the CRC stored with the message. In the event corruption is detected the log is truncated to the last valid offset”
สองเงื่อนไขนั้นจับความเสียหายคนละแบบ ความยาวจับ block ที่หายไปตอนเครื่องดับ ส่วน CRC จับ block มั่วที่ถูกเพิ่มเข้ามา ซึ่งเกิดได้เมื่อ inode ถูกอัปเดตก่อนเนื้อจะลงถึง
/// กฎของ Kafka: batch ใช้ได้ก็ต่อเมื่อ **ขนาดบวกตำแหน่งไม่เกินความยาวของ file**/// และ **CRC ตรง** · เจอตัวแรกที่ไม่ผ่านเมื่อไหร่ ให้ตัดทิ้งตั้งแต่ตรงนั้น////// ความเสียหายมีสองแบบ: ท้าย file ขาดเพราะ block ที่ยังไม่ถูกเขียนหายไปตอนเครื่องดับ/// และ block มั่วที่ถูก *เพิ่ม* เข้ามา · การตรวจ CRC จับแบบที่สอง การตรวจความยาวจับแบบแรกpub fn recover_truncate_tail(log: &mut Log) -> io::Result<Recovery> { let seg = log.segments.last_mut().expect("log มีอย่างน้อยหนึ่งท่อนเสมอ"); let file_len = seg.file_len()?; let mut pos = 0u64; let mut scanned = 0usize; let mut last_valid_offset = None;
while pos < file_len { let Some(header) = seg.read_batch_header_at(pos)? else { break }; let total = len_to_u64(header.total_size()); if pos + total > file_len { break; } let len = u32::try_from(header.total_size()).map_err(|_| bad("batch ยาวเกิน u32"))?; // `read_batch_at` ตรวจ CRC ให้ในตัว · batch ที่ CRC ไม่ตรงคือ block มั่ว if seg.read_batch_at(pos, len).is_err() { break; } scanned += 1; last_valid_offset = Some(header.last_offset()?); pos += total; }
let dropped = file_len - pos; if dropped > 0 { seg.truncate_to(pos)?; } if let Some(last) = last_valid_offset { log.next_offset = last + 1; log.high_watermark = log.next_offset; } Ok(Recovery { scanned_batches: scanned, last_valid_offset, valid_bytes: pos, dropped_bytes: dropped, })}demo ต่อครึ่ง batch ไว้ท้าย file แล้วเปิด log ขึ้นมาใหม่
== ท้าย file ที่ขาด · ตัดทิ้งถึง batch ที่ครบตัวสุดท้าย ==ก่อนทำให้เสีย 1278 byteต่อท้ายครึ่ง batch 147 byte → 1425 byteเปิดใหม่ · log end offset ที่อ่านได้ 12แต่ file ยังยาว 1425 byte · O_APPEND เขียนที่ท้าย file จริงไม่ใช่ที่ท้ายส่วนที่ครบ · ถ้าไม่ตัดทิ้ง batch ถัดไปจะไปนั่งหลังขยะscan ไป 6 batchoffset ที่ครบตัวสุดท้าย Some(11)เหลือไว้ 1278 byte · ตัดทิ้ง 147 byteความยาว file หลังตัด 1278เขียนต่อได้ที่ offset 12 · ไม่มีรูโหว่คั่นกลางKafka ใช้กฎเดียวกัน: ขนาดบวกตำแหน่งต้องไม่เกินความยาว file และ CRC ต้องตรงสองบรรทัดที่ต้องอ่านคู่กันคือบรรทัดที่บอก offset กับบรรทัดที่บอกความยาว การเปิด log ขึ้นมาใหม่คือการ replayreplayการอ่าน log ตั้งแต่ต้นจนจบเพื่อสร้างสถานะขึ้นใหม่ · broker ทำตอนเปิด เพื่อหาขอบของ segment และ offset ตัวถัดไปที่จะแจก คือเดิน log ตั้งแต่ต้นเพื่อสร้างสถานะขึ้นใหม่ และมันให้ offset ตัวถัดไปที่ถูกต้องแล้ว
แต่ความยาวของ file ยังเป็น 1425 byte เท่าเดิม เพราะการ replay อ่านอย่างเดียว ไม่ได้แก้อะไร
O_APPEND เขียนที่ท้าย file จริง ไม่ใช่ที่ท้ายส่วนที่ครบ
ถ้าไม่ตัดขยะทิ้งก่อน batch ถัดไปจะไปนั่งต่อจากขยะ และ log ก้อนนั้นก็จะอ่านไม่ได้ตั้งแต่จุดนั้นไป
การตัดหางเอาความยาวกลับมาที่ 1278 byte ซึ่งเป็นขอบของ batch ที่ครบตัวสุดท้าย แล้วการเขียนครั้งถัดไปได้ offset 12 ต่อจาก offset ที่ครบตัวสุดท้ายพอดี ไม่มีรูโหว่คั่นกลาง
คอร์ส kaen-kvstore บทที่ 6 ยิง SIGKILL ใส่ process จริงกลางการเขียน
แล้วดูว่าเปิดกลับมาได้ไหม เป็นวิธีที่ตรงกับความจริงที่สุดเท่าที่จะทำได้
บทนี้ไม่ได้ทำอย่างนั้น และเหตุผลไม่ใช่ความขี้เกียจ SIGKILL ทิ้งจำนวน byte ที่เขียนไปแล้ว
ไว้ไม่เท่ากันทุกรอบ ผลรันจึงเปลี่ยนทุกครั้งที่รัน
เทสต์ของคอร์สนี้รัน demo ห้ารอบแล้วบังคับให้ทั้งห้ารอบได้ผลเหมือนกันทุก byte demo จึงต่อครึ่ง batch เข้าไปเอง ซึ่งได้ท้าย file ที่ขาดแบบเดียวกัน แต่ขาดเท่าเดิมทุกรอบ
ของสองอย่างนี้พิสูจน์คนละข้อ SIGKILL พิสูจน์ว่าเครื่องดับจริงแล้วยังกู้ได้
ส่วนการต่อครึ่ง batch พิสูจน์ว่าตัวกู้คืนเจอขอบที่ถูกต้อง เครดิตของวิธีแรกเป็นของคอร์สที่ 22
ของที่บทนี้จงใจไม่ทำ
หัวข้อที่มีชื่อว่า “ของที่บทนี้จงใจไม่ทำ”หัวข้อนี้ประกาศเส้นขอบเขตของบท รวมถึงข้อที่ถอยหลังจากคอร์สก่อนโดยตั้งใจ
คอร์ส Async กับ Tokio จบด้วยการยกของขึ้น Tokio แล้วเถียงว่า task ต่อหนึ่งสายชนะ thread ต่อหนึ่งสาย ซึ่งเป็นข้อสรุปที่ยังถูกอยู่
คอร์สนี้ถอยลงมาที่ std ล้วนแบบ blocking โดยตั้งใจ เพราะเรื่องของบทนี้คือรูป byte บนสาย กับลำดับการลง disk ไม่ใช่ตัวจัดคิวงาน
การใส่ runtime แบบ async เข้ามาจะเพิ่มชั้นที่ต้องอธิบายอีกชั้น โดยไม่เปลี่ยนคำตอบของบทนี้สักข้อ ส่วนคำถามว่าอะไรเกิดขึ้นเมื่อคำขอถูกจอดไว้ เป็นของบทที่ 4
ของที่ยังไม่มีในบทนี้ และไม่มีบทไหนแอบอ้างว่ามี
- ไม่มีการนับ record ซ้ำ ส่งคำขอเดิมสองครั้งได้ record สองชุด ซึ่งเป็นของบทที่ 7
- ไม่มี compression standard library ไม่มี codec ให้ และ Kafka บีบทั้ง batch ไม่ใช่ทีละ record
- ไม่มี quota และไม่มีการยืนยันตัวตน client รายเดียวกิน broker ทั้งตัวได้
- ไม่มี replica ตัวที่สอง
acks=-1จึงยังตอบว่าไม่พร้อมเสมอ จนกว่าจะถึงบทที่ 8
บทที่ 4 จะกลับด้านของทางเดินนี้ คือคำขอที่ขอ record ที่ยังไม่มีใครเขียนมา และคำถามว่า broker ควรตอบว่าง หรือควรจอดคำขอนั้นไว้ก่อน
- Apache Kafka 4.3 — Protocol Guide
(ตรวจแล้ว 2026-09-03) — ประโยคเรื่องการอ่านความยาวสี่ byte แล้วอ่านต่ออีก N byte
ประโยคเรื่องเลข 16 bit สองตัวที่ระบุ schema ช่องของ request header v1
ประโยคเรื่องคำขอที่ค้างอยู่ได้ครั้งละหนึ่งอันต่อหนึ่งสาย และช่อง
base_offsetที่อยู่ระดับ partition ในคำตอบของ produce - RequestHeader.json tag 4.1.0
(ตรวจแล้ว 2026-09-03) —
"validVersions": "1-2"กับ"flexibleVersions": "2+"ซึ่งเป็นหลักฐานว่า header v2 มี tagged field ทั้งที่หน้า HTML ไม่ได้วาดไว้ - Apache Kafka 4.3 — Broker Configs
(ตรวจแล้ว 2026-09-03) — ค่าตั้งต้นของ
socket.request.max.bytesและค่าตั้งต้นของlog.flush.interval.messagesกับlog.flush.interval.ms - Apache Kafka 4.3 — Producer Configs
(ตรวจแล้ว 2026-09-03) — ค่าตั้งต้นของ
batch.sizeกับlinger.msประโยคที่ว่าชุดเต็มแล้วส่งทันทีโดยไม่สนค่าหน่วง ซึ่งอยู่ในคำอธิบายของlinger.msประโยคที่ว่าขนาดชุดเป็นศูนย์คือปิด batching ประโยคที่เทียบตัวเองกับ Nagle’s algorithm และประโยคใต้acks=0ที่ว่า offset ซึ่งคืนให้ทุก record จะเป็นลบหนึ่งเสมอ - ReplicaManager.scala tag 4.1.0
(ตรวจแล้ว 2026-09-03) —
isValidRequiredAcksที่รับแค่ลบหนึ่ง หนึ่ง และศูนย์ กับคอมเมนต์ที่บอกว่าค่าที่หลุดช่วงแปลว่า client ผิด - KafkaApis.scala tag 4.1.0
(ตรวจแล้ว 2026-09-03) — คอมเมนต์ที่บรรทัด 501 ซึ่งบอกว่าเมื่อ producer ไม่ได้รอคำตอบอยู่
แล้วเกิด error ขึ้น server จะปิด socket แทนการตอบ กับ
requestChannel.closeConnectionที่บรรทัด 512 - UnifiedLog.java tag 4.1.0
(ตรวจแล้ว 2026-09-03) — บรรทัด
if (localLog.unflushedMessages() >= config().flushInterval) flush(false);ที่บรรทัด 1248 และคอมเมนต์เรื่องการเลี่ยง fsync ตอนถือUnifiedLog#lockที่บรรทัด 2109 - Apache Kafka 4.3 — Design (ตรวจแล้ว 2026-09-03) — ประโยค “two to three orders of magnitude” ที่ไม่มี benchmark ไม่มีเครื่อง และไม่มีวันที่แนบมาด้วย และประโยคที่ว่าข้อมูลลง pagecache ของ kernel ก่อน
- Apache Kafka 4.3 — Hardware and OS (ตรวจแล้ว 2026-09-03) — ประโยคที่แนะนำให้ใช้ค่าตั้งต้นซึ่งปิด fsync ฝั่ง application ทิ้ง และประโยคที่ว่าการรับประกันจากการทำสำเนาแข็งแรงกว่าการ sync ลง disk ของเครื่องตัวเอง
- Apache Kafka 4.0 — Implementation › Log (ตรวจแล้ว 2026-09-03) — กฎว่า entry ใช้ได้เมื่อขนาดบวกตำแหน่งไม่เกินความยาว file และ CRC ตรง กับความเสียหายสองแบบที่ต้องรับมือ
- Rust std — trait Read
(ตรวจแล้ว 2026-09-03) — ประโยคที่ว่า
readคืน byte ไม่ครบได้ทั้งที่ยังไม่จบ stream และประโยคของread_exactที่ว่าเมื่อคืน error แล้วจะไม่ระบุว่าอ่านไปกี่ byte - Rust std — struct File
(ตรวจแล้ว 2026-09-03) — คำอธิบายของ
sync_allรวมถึงประโยคที่ว่าการปล่อยFileทิ้ง จะกลืน error ทุกตัว
เช็กความเข้าใจ — บทที่ 3
ข้อ 1 / 3kaen-log เรียก `sync_all` ก่อนตอบทุกครั้ง ส่วน Kafka ไม่ fsync บนทาง write เลย ข้อไหนอธิบายความต่างนี้ได้ถูกต้อง