เลขสองตัวของผู้บริโภค · ตำแหน่ง กับ offset ที่ commit แล้ว

บทที่ 4 วางทางอ่านไว้ครบแล้ว ผู้อ่านขอ record จาก offset ที่ต้องการ แล้วได้ของกลับไป สิ่งที่ยังไม่มีใครตอบคือ ผู้อ่านรู้ได้อย่างไรว่าครั้งหน้าต้องขอจากตรงไหน

คำตอบคือมันไม่ได้ถือเลขตัวเดียว มันถือสองตัว ตัวหนึ่งอยู่ในหน่วยความจำของมันเอง อีกตัวอยู่บน disk ของ broker และมีแค่ตัวหลังที่รอดจากการล้ม

ผลต่างของสองตัวนั้นคือจำนวน order ที่จะถูกทำซ้ำหลัง restart และลำดับของสองบรรทัดใน loop เป็นตัวตัดสินว่าจะได้การทำซ้ำ หรือได้ order ที่หายไปเลย

📦 code ของบทนี้

crate ของคอร์สอยู่ที่ scripts/log-broker/ และชื่อ binary คือ klog เหมือนทุกบท

บล็อก code กับบล็อกผลรันทุกอันในบทเรียนนี้ถูกคัดมาจาก scripts/log-broker/src/group.rs และจาก scripts/log-broker/expected/05-group.txt ทีละ byte โดยมีเทสต์เทียบไว้ ถ้าแก้ในบทเรียนอย่างเดียวโดยไม่แก้ที่ต้นทาง เทสต์จะแดงทันที

ผลรันทุกบรรทัดในบทนี้มาจากคำสั่งเดียว คือ klog group-demo --dir data --group kitchen --crash-after 12 --now-ms 1700000000000 ซึ่งเทสต์รันซ้ำแล้วเทียบผลทีละ byte ทุกครั้ง

บทนี้เป็นบทแรกที่สร้าง partition ใบที่สอง คือ data/__offsets-0/ demo ลบ folder ทั้งสองใบทิ้งทั้งก่อนและหลังรัน

ผู้อ่านหนึ่งรายถือเลขสองตัว และมีตัวเดียวที่รอดจากการล้ม

หัวข้อนี้แยกตำแหน่งในหน่วยความจำ ออกจาก offset ที่ commit แล้วบน disk

ผู้อ่านของ kaen-log ไม่ได้ทำงานคนเดียว มันสังกัด consumer group หนึ่งกลุ่ม และ log ของคอร์สนี้มีสามกลุ่มที่อ่านอยู่พร้อมกัน คือ kitchen courier และ analytics

เลขตัวแรกที่สมาชิกกลุ่มถือคือ position มันคือ offset ตัวถัดไปที่จะขอ มันเดินหน้าเองทุกครั้งที่ poll ได้ของกลับมา และมันอยู่ในหน่วยความจำของ process นั้นล้วนๆ

เอกสารของ KafkaConsumer เขียนนิยามนี้ไว้เอง ว่าตำแหน่ง “gives the offset of the next record that will be given out. It will be one larger than the highest offset the consumer has seen in that partition. It automatically advances every time the consumer receives messages in a call to poll(Duration)”

เลขตัวที่สองคือ committed offset เอกสารหน้าเดียวกันเรียกมันว่า committed position และนิยามว่า “is the last offset that has been stored securely. Should the process fail and restart, this is the offset that the consumer will recover to”

คำที่แยกสองเลขนี้ออกจากกันคือ securely ตำแหน่งไม่ได้ถูกเก็บที่ไหนเลย ส่วน offset ที่ commit แล้วถูกเขียนลง log ของ broker และรอดจากการที่ process ตาย

ประโยคปิดของย่อหน้านั้นคือเหตุผลที่คอร์สนี้ให้ทั้งบทกับเรื่องนี้ ว่า “This distinction gives the consumer control over when a record is considered consumed”

🔁 คอร์สที่ 22 ไม่มีเลขตัวนี้เลยสักตัว

คอร์ส kaen-kvstore เก็บ log ที่ต่อท้ายอย่างเดียวไว้แล้ว มีสารบัญของมันเอง มี compaction ของมันเอง และมีการกู้คืนหลัง crash ของมันเอง

สิ่งที่มันไม่มีคือความคิดเรื่อง ตำแหน่งของผู้อ่าน เพราะผู้อ่านของมันคือตัวมันเอง get หนึ่งครั้งเปิดสารบัญแล้ว seek ไปอ่าน ไม่มีใครเดินไปข้างหน้าและไม่มีใครต้องจำว่าอ่านถึงไหน

ช่องว่างนั้นคือสิ่งที่บทนี้เติม และมันเป็นช่องว่างที่โผล่ขึ้นมาตอนที่ผู้อ่านกลายเป็นคนแปลกหน้า ที่อยู่คนละ process กับคนเขียน

offset ที่ commit คือ record ตัวถัดไป ไม่ใช่ตัวสุดท้ายที่ทำเสร็จ

หัวข้อนี้ตรึงความหมายของค่าที่เก็บ ว่าเป็นเลขของ record ตัวถัดไป

ค่าที่ commit ลงไปไม่ใช่หมายเลขของ record ที่เพิ่งประมวลผลเสร็จ มันคือหมายเลขของ record ตัวถัดไป ประมวลผล record หมายเลข N เสร็จแล้วต้องเก็บ N+1

src/group.rs เขียนข้อนี้ไว้ที่หัวของ struct เอง เพราะมันคือที่เดียวที่ผิดแล้วจะไม่มีอะไรเตือน

/// **`offset` คือ record ตัวถัดไปที่จะอ่าน ไม่ใช่ตัวสุดท้ายที่ประมวลผลไปแล้ว**
/// · commit หลังประมวลผล record หมายเลข N แปลว่าเก็บค่า N+1
///
/// บน wire ค่านี้เป็น `int64` · ในหน่วยความจำเป็น `u64` และ `from_wire_i64`
/// เป็นตัวเดียวที่แปลง โดยปฏิเสธค่าติดลบ
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OffsetValue {
    pub offset: u64,
    pub leader_epoch: i32,
    pub metadata: String,
    pub commit_timestamp_ms: i64,
}

เอกสารของ KafkaConsumer เน้นข้อนี้ด้วยตัวหนาในหน้าของมันเอง ว่า “The committed offset should always be the offset of the next message that your application will read”

ประโยคเดียวกันในเอกสารของ commitSync เติมทางที่ปลอดภัยกว่าเข้าไปด้วย ว่าให้ใช้ nextRecordToBeProcessed.offset() หรือ ConsumerRecords#nextOffsets() แทนการบวกหนึ่งเอง

ข้อเท็จจริงที่ทำให้การบวกหนึ่งเองน่ากลัวอยู่ในเอกสารหน้าเดียวกัน ว่า offset “are not guaranteed to be consecutive (such as compacted topic or when records have been produced using transactions)” เอกสารไม่ได้ผูกสองข้อนี้เข้าด้วยกันเอง คอร์สนี้เป็นคนผูก

บทที่ 6 คือบทที่สร้างช่องว่างแบบนั้นขึ้นมาจริงๆ compaction ลบ record ทิ้งได้โดยไม่ขยับ offset ของตัวที่เหลือ ลำดับที่เคยติดกันจึงขาดเป็นช่วงๆ ได้

run_at_least_once ของ kaen-log ก็บวกหนึ่งเองเหมือนกัน สิ่งที่ทำให้มันรอดช่องว่างไม่ใช่การบวก แต่เป็น poll ที่คัดเฉพาะ record ที่ offset >= from ค่าที่ตกลงไปในช่องว่างจึงได้ตัวถัดไป ที่ยังมีอยู่จริง เอกสารของ Kafka เขียนพฤติกรรมเดียวกันไว้ ว่าถ้าตำแหน่งเป็น 5 แต่ไม่มี record ที่ offset 5 “the consumer will return the record with the next higher offset”

หน้าต่างของการส่งซ้ำ คือผลลบของเลขสองตัว

หัวข้อนี้ล้ม consumer กลางคัน แล้ววัดว่ามี order กี่ตัวที่ถูกทำซ้ำ

ลำดับของ at-least-once คืออ่าน แล้วประมวลผล แล้วค่อยเก็บตำแหน่ง เอกสารออกแบบของ Kafka เขียนลำดับนี้ไว้ตรงตัว ว่า “It can read the messages, process the messages, and finally save its position”

/// อ่าน → ประมวลผล → แล้วค่อยเก็บตำแหน่ง · **at-least-once**
pub fn run_at_least_once(
    orders: &Partition,
    offsets: &Partition,
    group: &str,
    start: u64,
    crash_after: u64,
    poll_size: usize,
    now_ms: i64,
) -> io::Result<RunOutcome> {
    let mut position = start;
    let mut committed = start;
    let mut processed = Vec::new();
    while position < orders.high_watermark() {
        let records = poll(orders, position, poll_size)?;
        if records.is_empty() {
            break;
        }
        for record in &records {
            processed.push(record.offset);
            position = record.offset + 1;
            if u64::try_from(processed.len()).map_err(|_| bad("นับเกิน u64"))? >= crash_after
            {
                return Ok(RunOutcome { position, committed, processed });
            }
        }
        committed = position;
        commit(offsets, group, committed, now_ms)?;
    }
    Ok(RunOutcome { position, committed, processed })
}

position ขยับทุก record ส่วน committed ขยับเมื่อจบชุดของ poll หนึ่งครั้งเท่านั้น บรรทัด return ตรงกลางคือการล้ม มันออกจาก function โดยไม่ผ่านสองบรรทัดล่างสุดของ loop

demo สั่งให้มันล้มหลังประมวลผลครบ 12 ตัว โดย poll ครั้งละ 5 record

== at-least-once · อ่าน → ประมวลผล → เก็บตำแหน่ง ==
group kitchen · poll ครั้งละ 5 record · ล้มหลังประมวลผลครบ 12 ตัว
ตำแหน่ง (ในหน่วยความจำ)   12
offset ที่ commit (บน disk)  10
หน้าต่างการส่งซ้ำ            2 = 12 - 10
restart แล้วอ่าน offset จาก log ของ broker ได้ 10
record ที่ถูกประมวลผลซ้ำ      2 ตัว · offset [10, 11]
เท่ากับหน้าต่างพอดี · ไม่มี order หาย มีแต่ order ที่ทำซ้ำ

ตอนล้ม ตำแหน่งอยู่ที่ 12 ส่วนค่าที่ commit ไว้บน disk ยังเป็น 10 ผลลบของสองตัวนี้คือ 2 และมันไม่ใช่การประมาณ มันคือจำนวนที่จะถูกทำซ้ำจริง

บรรทัดถัดมาพิสูจน์ข้อนั้นด้วยการรันจริง ไม่ใช่ด้วยการคำนวณ demo restart ผู้อ่านขึ้นมาใหม่ อ่านค่าที่ commit ไว้จาก log ของ broker ได้ 10 แล้วเดินต่อ record ที่ตรงกับรอบแรกมีสองตัวพอดี คือ offset 10 กับ 11

demo ตัวนี้ยังมีด่านของตัวเองอยู่ข้างใน ถ้าจำนวนที่ทำซ้ำจริงไม่เท่ากับผลลบที่มันพิมพ์ไป มันจะ error ออกมาแทนที่จะพิมพ์ต่อ ข้ออ้างที่วัดไม่ได้จึงไม่มีทางหลุดออกมาเป็นบรรทัดสวยๆ

sequenceDiagram
    participant C as ผู้อ่านของ group kitchen
    participant B as partition orders-0
    participant O as log ของ __offsets
    C->>B: fetch จากตำแหน่ง 10
    B-->>C: record 10 ถึง 14
    Note over C: ทำ order 10 กับ 11 เสร็จ · ตำแหน่งเดินไปที่ 12
    C--xC: ล้มก่อนถึงบรรทัด commit
    Note over O: ค่าที่เก็บไว้ล่าสุดยังเป็น 10
    C->>O: restart แล้วขอค่าที่ commit ไว้
    O-->>C: 10
    C->>B: fetch จากตำแหน่ง 10 อีกครั้ง
    Note over C: order 10 กับ 11 ถูกทำซ้ำ · เท่ากับ 12 ลบ 10

คำบรรยายภาพ: ลูกศรที่ย้อนกลับมาที่ตัวเองคือการล้ม ตำแหน่งที่เดินไปถึง 12 อยู่ในหน่วยความจำของ process ที่ตายไปแล้ว ส่วนค่าที่ผู้อ่านคนใหม่หยิบขึ้นมาได้คือค่าที่ commit ไว้บน disk เท่านั้น ระยะระหว่างสองค่านั้นคือ order ที่ถูกทำซ้ำ

สลับสองบรรทัด แล้ว order หายจริง ไม่ใช่แค่ทำซ้ำ

หัวข้อนี้ย้าย commit ขึ้นไปก่อนการประมวลผล แล้วนับ order ที่หายไป

เอกสารออกแบบของ Kafka เขียนอีกทางเลือกไว้ก่อนหน้านั้นหนึ่งข้อ ว่า “It can read the messages, then save its position in the log, and finally process the messages” และสรุปผลของมันว่า “This corresponds to ‘at-most-once’ semantics as in the case of a consumer failure messages may not be processed”

run_at_most_once ใน crate ต่างจาก function ข้างบนที่เดียว คือการ commit ย้ายขึ้นมาอยู่ก่อน loop ประมวลผล และเพราะตำแหน่งยังไม่ได้เดิน มันจึงต้องคิดค่าที่จะ commit จาก record ตัวสุดท้ายของชุดแทน

        let last = records.last().expect("ตรวจว่าไม่ว่างไปแล้ว").offset;
        committed = last + 1;
        commit(offsets, group, committed, now_ms)?;
        for record in &records {

การ commit เกิดขึ้นก่อนที่ record ชุดนั้นจะถูกแตะเลยสักตัว ผลของการสลับที่พิมพ์ออกมาแบบนี้

== at-most-once · อ่าน → เก็บตำแหน่ง → ประมวลผล ==
ตำแหน่ง (ในหน่วยความจำ)   12
offset ที่ commit (บน disk)  15
record ที่หายไปเลย           3 ตัว · offset 12 ถึง 14
สลับสองบรรทัดแล้ว order หายจริง ไม่ใช่แค่ทำซ้ำ
นี่คือความต่างทั้งหมดระหว่างสองความหมาย · ไม่มีอะไรอื่นเลย

คราวนี้ค่าที่ commit ไว้เดินนำตำแหน่งจริง ผู้อ่านคนใหม่จะเริ่มที่ 15 ส่วน order หมายเลข 12 ถึง 14 ไม่เคยถูกประมวลผล และไม่มีใครกลับไปหยิบมันอีก

ลำดับตัดสินว่าจะพังทางไหน ส่วนความถี่ของการ commit ตัดสินว่าพังกว้างแค่ไหน ในการรันคู่นี้ระยะทั้งสองบวกกันได้ 5 พอดี ซึ่งเท่ากับขนาดของ poll หนึ่งครั้ง เพราะจุดที่ล้มอยู่กลาง poll ชุดเดียวกัน

⚠ ค่าปริยายของ client เลือกข้างให้แล้ว และมันเลือกทางที่มีของหายได้

enable.auto.commit ของ KafkaConsumer มีค่าปริยายเป็น true ผู้อ่านที่ไม่ได้ตั้งอะไรเลยจึง commit ให้เองเป็นระยะอยู่เบื้องหลัง

เอกสารของมันเตือนไว้ตรงตัวว่า auto commit ให้ at-least-once ก็ต่อเมื่อทำครบเงื่อนไข “you must consume all data returned from each call to poll(Duration) before any subsequent calls” และถ้าไม่ครบ “it is possible for the committed offset to get ahead of the consumed position, which results in missing records”

นั่นคือรูปเดียวกับ at-most-once เป๊ะ คือค่าที่ commit เดินนำตำแหน่งจริง มันไม่ได้มาจากการสลับบรรทัดด้วยมือ แต่มาจากการปล่อยให้ตัว commit เดินตามนาฬิกาของมันเอง

offset ที่ commit ไปนั่งอยู่ใน log ของ broker เอง

หัวข้อนี้เปิด partition ใบที่สอง แล้วบอกว่า key ของ record ในนั้นคืออะไร

broker ไม่ได้เก็บ offset ที่ commit ไว้ในตารางพิเศษ มันเก็บเป็น record ธรรมดาใน log ของตัวเอง คอร์สนี้ตั้งชื่อ partition นั้นว่า __offsets

/// ชื่อ partition ภายในของคอร์สนี้ · ของ Kafka ชื่อ `__consumer_offsets`
/// และมี **50 partition กับ replication factor 3** · ของ kaen-log มี partition เดียว
/// และ **ไม่มี replica เลย** · บทเรียนต้องพูดข้อนี้ ไม่ใช่ปล่อยผ่าน
pub const OFFSETS_TOPIC: &str = "__offsets";

ของ Kafka ชื่อ __consumer_offsets และ GroupCoordinatorConfig ตั้งค่าปริยายไว้ที่ 50 partition กับ replication factor 3 ส่วนคำอธิบายของค่าแรกเขียนกำกับไว้ด้วยว่า “should not change after deployment”

kaen-log มี partition เดียวและไม่มีสำเนาของมันเลยสักใบ นี่ไม่ใช่รายละเอียดเล็กน้อย มันแปลว่า disk ใบที่พัง พา offset ที่ commit ไว้ของทั้งสามกลุ่มไปด้วยพร้อมกัน

== partition สองใบ ==
  folder                  topic          cleanup.policy
  data/orders-0            orders         delete
  data/__offsets-0         __offsets      compact
log end offset ของ orders 20
`__offsets` ของ Kafka ชื่อ `__consumer_offsets` มี 50 partition และ RF 3
ของ kaen-log มี partition เดียว และไม่มี replica เลยสักตัว

สอง partition นี้ต่างกันที่ cleanup.policy ใบของผู้ใช้ตั้งเป็น delete ส่วนใบของ broker ตั้งเป็น compact และของ Kafka ก็ตั้งแบบเดียวกัน GroupCoordinatorService.groupMetadataTopicConfigs ใส่ TopicConfig.CLEANUP_POLICY_COMPACT ให้ topic นั้นตอนสร้าง

key ของ record คือ triple พอดี

key ต้องบอกให้ครบว่าใครอ่านอะไรถึงไหน และครบพอดีแปลว่าไม่มีอะไรเกิน

/// key คือ triple พอดี · ไม่มีอะไรเกินและไม่มีอะไรขาด
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct OffsetKey {
    pub group: String,
    pub topic: String,
    pub partition: i32,
}

รูปของ key ลอกมาจาก OffsetCommitKey.json ของ Kafka ทีละช่อง schema นั้นประกาศ validVersions ไว้ที่ "0-1" และมีสาม field พอดี คือ group เป็น string topic เป็น string และ partition เป็น int32

ที่ลอกมาคือรายชื่อของ field ไม่ใช่การ encode ตัว kaen-log เขียนสาม field นี้ลง byte ด้วยวิธีของมันเอง และไม่ได้เข้ากันได้ทีละ byte กับ client ของ Kafka ตามที่บทที่ 1 ประกาศไว้ตั้งแต่ต้น

การที่ key มี group อยู่ข้างในคือเหตุผลที่สามกลุ่มอยู่คนละตำแหน่งบน log เดียวกันได้ มันไม่ใช่ feature ที่เติมเข้ามา มันเป็นผลของการเลือก key

commit หนึ่งครั้งคือ produce หนึ่งครั้ง

append_to_offsets_log ไม่ได้เขียนลง log ด้วยทางลัดของตัวเอง มันเรียกทาง produce เส้นเดียวกับที่ บทที่ 3 สร้างไว้ จึงได้การแจก offset การ fsync และการตอบกลับแบบเดียวกันทั้งหมด

    let record = Record {
        offset: 0,
        timestamp_ms: now_ms,
        key: Some(key.encode()),
        // value เป็น `None` คือ tombstone · เส้นทางเดียวกับที่คำสั่งลบใช้
        value: value.map(OffsetValue::encode),
    };

ช่อง offset เป็นศูนย์ตอนส่ง เพราะบทที่ 1 ตรึงข้อนี้ไว้แล้วว่า broker เป็นคนแจกลำดับ ไม่ใช่ผู้ส่ง แม้ผู้ส่งครั้งนี้จะเป็นตัว broker เองก็ตาม

replay ของ `__offsets` และ tombstone ที่บทที่ 6 จะเก็บกวาด

หัวข้อนี้อ่าน log ของ offset ทั้งก้อนกลับมาเป็นตาราง แล้วลบ key หนึ่งตัวทิ้ง

__offsets เป็น log ที่ต่อท้ายอย่างเดียวเหมือนกับ orders ทุกประการ การจะรู้ว่าตอนนี้แต่ละกลุ่มอ่านถึงไหน จึงต้องอ่านทั้ง log แล้วเก็บค่าสุดท้ายของแต่ละ key

/// replay ทั้ง log แล้วเก็บค่าสุดท้ายต่อ key · tombstone ลบ key นั้นออก
///
/// ใช้ `BTreeMap` ไม่ใช่ `HashMap` เพราะ demo พิมพ์ตารางนี้ออกมา และลำดับของ
/// `HashMap` ไม่แน่นอนระหว่างการรันแต่ละครั้ง
pub fn load_offsets(offsets: &Partition) -> io::Result<BTreeMap<OffsetKey, OffsetValue>> {
    let mut out = BTreeMap::new();
    let log = offsets.log_guard();
    for segment in log.segments() {
        for (_, batch) in segment.read_all_batches()? {
            for record in batch.records {
                // record ที่ไม่มี key ผิดกฎบน topic ที่ compact · Kafka ปฏิเสธตั้งแต่ produce
                let Some(raw_key) = record.key.as_deref() else { continue };
                let key = OffsetKey::decode(raw_key)?;
                match record.value.as_deref() {
                    Some(raw) => {
                        out.insert(key, OffsetValue::decode(raw)?);
                    }
                    None => {
                        out.remove(&key);
                    }
                }
            }
        }
    }
    Ok(out)
}

สอง branch ล่างสุดคือทั้งหมดของความหมาย value ที่มีของแปลว่าทับค่าเดิม ส่วน value ที่เป็น None แปลว่าลบ key นั้นออกจากผล ซึ่งก็คือ tombstone ที่บทที่ 1 นิยามไว้แล้ว ว่าเป็นความยาวลบหนึ่งบน wire

การลบจึงไม่ใช่การไปแก้ของเก่า มันคือการต่อท้าย record ใหม่หนึ่งตัวที่ไม่มี value

== ลบ offset ของกลุ่มหนึ่งทิ้ง · tombstone เส้นทางเดียวกับบทที่ 6 ==
key ที่ยังมีชีวิตก่อนลบ 3 · หลังลบ 2
log ยาวขึ้นหนึ่ง record แต่ key ที่ยังมีชีวิตน้อยลงหนึ่งตัว
log end offset ของ __offsets 7
บทที่ 6 เป็นคนเก็บกวาด record พวกนี้ ด้วยตัวทำความสะอาดตัวเดียวกับ orders

สองบรรทัดกลางคือความย้อนแย้งที่ทำให้ต้องมีบทที่ 6 log ยาวขึ้นทุกครั้งที่มีคนลบของ ส่วนจำนวนของที่ยังมีชีวิตกลับน้อยลง

บทที่ 6 · retention กับ compaction คือคนที่มาเก็บกวาดส่วนต่างนั้น และมันใช้ตัวทำความสะอาดตัวเดียวกับที่ใช้กับ orders ไม่มีกรณีพิเศษให้ partition ภายใน

สามกลุ่ม อ่าน log เดียวกัน อยู่คนละตำแหน่ง

หัวข้อนี้ commit ให้สามกลุ่มที่สามค่า แล้วอ่านทั้งตารางกลับออกมา

demo เขียน commit ของสามกลุ่มลง __offsets ทีละตัว แล้วอ่านทั้ง log กลับมาเป็นตาราง

== สามกลุ่ม อ่าน log เดียวกัน ที่สาม offset ==
  commit ของ kitchen    เก็บที่ offset 3 ของ __offsets
  commit ของ courier    เก็บที่ offset 4 ของ __offsets
  commit ของ analytics  เก็บที่ offset 5 ของ __offsets
  group        topic    partition   offset ถัดไปที่จะอ่าน
  analytics    orders           0   0
  courier      orders           0   4
  kitchen      orders           0   7
การอ่านไม่กินของใคร · สามกลุ่มอยู่คนละตำแหน่งบน log เดียวกันได้ตลอดไป

สามแถวบนคือ record สามตัวใน __offsets เรียงตามลำดับที่เขียนลงไป สามแถวล่างคือผลของการ replay ซึ่งเรียงตามชื่อกลุ่ม เพราะ BTreeMap เรียงตาม key ให้

ตัวเลขที่น่าสนใจอยู่ท้ายแถว analytics ยังอยู่ที่ 0 คือยังไม่ได้เริ่มอ่าน courier อยู่ที่ 4 และ kitchen อยู่ที่ 7 ทั้งสามอ่าน log ก้อนเดียวกันอยู่

เอกสารออกแบบของ Kafka อธิบายว่าทำไมสิ่งนี้ถึงถูก โดยเทียบกับระบบคิวที่เก็บสถานะไว้ที่ broker ทีละข้อความ ระบบแบบนั้น “must keep multiple states about every single message” คือทั้งการล็อกไม่ให้แจกซ้ำ และการมาร์คว่าบริโภคแล้ว

kaen-log ไม่เก็บสถานะรายข้อความสักตัว มันเก็บจำนวนเต็มหนึ่งตัวต่อหนึ่ง triple และการอ่านไม่ทำให้ของหายไปจากใคร

หนึ่ง partition ต่อหนึ่งสมาชิกในกลุ่มเดียวกัน

ประโยคที่เอกสารออกแบบเขียนไว้คือ topic ถูกแบ่งเป็น partition “each of which is consumed by exactly one consumer within each subscribing consumer group at any given time”

ประโยคเดียวกันบอกต่อว่าผลของมันคือ “the position of a consumer in each partition is just a single integer, the offset of the next message to consume” และปิดท้ายว่า “This makes the equivalent of message acknowledgements very cheap”

ข้อที่ตามมาจากประโยคนั้นคือเพดานของการขยาย ถ้ากลุ่มหนึ่งมีสมาชิกมากกว่าจำนวน partition สมาชิกที่เกินมาจะไม่มี partition ให้ถือ

ข้อสรุปนั้นเป็นการอนุมานจากกติกาข้างบน ไม่ใช่ประโยคที่เอกสารของ Kafka เขียนไว้เอง คอร์สนี้ค้นแล้วไม่เจอประโยคแบบนั้น จึงเสนอมันในฐานะข้อสรุป ไม่ใช่ในฐานะคำอ้าง

กันผีด้วยจำนวนเต็มตัวเดียว

หัวข้อนี้ให้สมาชิกรุ่นเก่าลอง commit แล้วดูว่าอะไรปฏิเสธมัน

สมาชิกที่ค้างอยู่จากการจัดสมาชิกรอบก่อนคือปัญหาจริง มันยังถือ partition ในความคิดของตัวเอง และยังพยายาม commit ทับค่าที่กลุ่มปัจจุบันเพิ่งเขียนไป

เครื่องมือที่ใช้กันมันคือ generation fencing ซึ่งเป็นจำนวนเต็มตัวเดียวที่เพิ่มขึ้นทุกครั้ง ที่กลุ่มจัดสมาชิกใหม่

/// สมาชิกเก็บเป็น `Vec` ไม่ใช่ `HashMap` · ลำดับของ `HashMap` ไม่แน่นอน และ demo
/// ตัวนี้พิมพ์รายชื่อสมาชิกออกมา
pub struct Group {
    pub id: String,
    pub state: GroupState,
    /// การกันผีเป็น `i32` ตัวเดียว · ตรงกับ `GenerationIdOrMemberEpoch` บน wire
    pub generation: i32,
    pub members: Vec<String>,
    pub last_heartbeat_ms: Vec<(String, i64)>,
}

ความกว้างของช่องนั้นไม่ใช่รสนิยม OffsetCommitRequest.json ของ Kafka ประกาศ field ชื่อ GenerationIdOrMemberEpoch เป็น int32 มีค่าปริยายเป็น -1 และคำอธิบายของมันบอกว่ามันคือ “The generation of the group if using the classic group protocol or the member epoch if using the consumer protocol”

ตัวตัดสินใจของ kaen-log อ่านสถานะก่อน แล้วค่อยอ่านเลขรุ่น

    /// **การ rebalance ปิดการเขียน ไม่ได้ปิดการอ่าน** · `PREPARING_REBALANCE`
    /// ยังรับ commit จาก generation ก่อนหน้าได้ ส่วน `COMPLETING_REBALANCE` ไม่รับ
    /// · การ **อ่าน** offset ทำได้ในทุกสถานะที่ยังมีชีวิต
    pub fn check_commit(&self, member: &str, generation: i32) -> Result<(), GroupError> {
        match self.state {
            GroupState::CompletingRebalance => return Err(GroupError::RebalanceInProgress),
            GroupState::Dead => return Err(GroupError::UnknownMemberId),
            _ => {}
        }
        if generation != self.generation {
            return Err(GroupError::IllegalGeneration);
        }
        if !self.members.iter().any(|m| m == member) {
            return Err(GroupError::UnknownMemberId);
        }
        Ok(())
    }

คำอธิบายเหนือ function ตัวนี้เขียนกฎของ Kafka ไว้ ไม่ใช่กฎที่ code ข้างล่างมันบังคับครบทุกข้อ ส่วนที่ไม่ตรงกันอยู่ใน callout ท้ายหัวข้อนี้

รูปของ match ลอกมาจาก ClassicGroupState ของ Kafka ซึ่งเขียนสิ่งที่แต่ละสถานะทำไว้เป็นรายการ COMPLETING_REBALANCE เขียนว่า “respond to offset commits with REBALANCE_IN_PROGRESS” ส่วน DEAD เขียนว่า “respond to offset commit with UNKNOWN_MEMBER_ID”

demo ยิงหกกรณีเข้าไปที่ด่านนี้ แล้วพิมพ์ผลของทั้งการเขียนและการอ่านคู่กัน

== กันผีด้วย i32 ตัวเดียว ==
group kitchen · สถานะ Stable · generation 2 · สมาชิก ["member-a", "member-b"]
group ที่ยังไม่มีสมาชิกเลยอยู่ในสถานะ Empty
  กรณี                            commit             fetch
  สมาชิกจริง generation ปัจจุบัน  ผ่าน               ผ่าน
  ผีจาก generation ก่อน           ILLEGAL_GENERATION  ผ่าน
  สมาชิกที่ไม่รู้จัก              UNKNOWN_MEMBER_ID  ผ่าน
  กำลังเตรียม rebalance           ผ่าน               ผ่าน
  กำลังปิด rebalance              REBALANCE_IN_PROGRESS  ผ่าน
  group ตายแล้ว                   UNKNOWN_MEMBER_ID  UNKNOWN_MEMBER_ID
rebalance ปิดการ **เขียน** ไม่ได้ปิดการ **อ่าน**
PreparingRebalance ยังรับ commit จาก generation ปัจจุบันได้ · CompletingRebalance ไม่รับ
การอ่าน offset ผ่านได้ทุกสถานะที่ยังมีชีวิต · ตกเฉพาะตอน group ตายแล้ว

column ขวาสุดคือข้อที่คนมองข้ามบ่อยที่สุด การอ่านผ่านทุกแถวยกเว้นแถวสุดท้าย การ rebalance ปิดการเขียน ไม่ได้ปิดการอ่าน

ClassicGroupState พูดข้อนี้ด้วยความสม่ำเสมอที่หายาก ทั้งห้าสถานะเขียนบรรทัด “allow offset fetch requests” ไว้เหมือนกันหมด รวมทั้ง DEAD ที่ปฏิเสธการ commit ไปแล้ว

และ kaen-log ไม่ได้ตามข้อสุดท้ายนั้น แถวล่างสุดของตารางแสดงว่าการอ่านตกด้วยตอนที่ group ตายแล้ว ซึ่งต่างจาก Kafka ที่ยังยอมให้อ่าน ตารางเดียวกันที่ใช้แสดงกฎ จึงฟ้องข้อที่ kaen-log ไม่ได้ทำตามด้วย

kaen-log เข้มกว่า Kafka ตรงเลขรุ่น และข้อนั้นสังเกตได้จากตารางเดียวกัน

PREPARING_REBALANCE ของ Kafka เขียนว่า “allow offset commits from previous generation” คือมันยอมรับ commit จากรุ่น ก่อนหน้า ระหว่างที่กำลังเตรียมจัดสมาชิกใหม่

check_commit ของ kaen-log ไม่ทำแบบนั้น มันเทียบเลขรุ่นเท่ากันเป๊ะในทุกสถานะที่ยังมีชีวิต แถว “กำลังเตรียม rebalance” ในตารางข้างบนจึงผ่าน เพราะกรณีนั้นใช้ generation ปัจจุบัน ไม่ใช่เพราะ kaen-log ยอมรับรุ่นเก่า

ความต่างข้อนี้ทำให้ kaen-log ปฏิเสธ commit ที่ Kafka จะรับ และผลคือผู้อ่านที่กำลังจะถูกจัดใหม่ เสีย commit สุดท้ายไป ซึ่งขยายหน้าต่างการส่งซ้ำให้กว้างขึ้น ไม่ได้ทำให้ order หาย

ราคาที่การ rebalance เก็บจากทั้งกลุ่ม

หัวข้อนี้บอกว่าการจัดสมาชิกใหม่หยุดอะไร และคอร์สนี้ไม่มีตัวเลขอะไรให้

การจัดสมาชิกใหม่ไม่ได้แพงเพราะการคำนวณว่าใครถือ partition ไหน มันแพงเพราะ ช่วง CompletingRebalance ปิด commit ของทั้งกลุ่มไว้ ตารางข้างบนแสดงแถวนั้นไว้แล้ว ส่วนแถว PreparingRebalance ยังผ่าน ราคาจึงเก็บเฉพาะช่วงหลัง ไม่ใช่ตลอดการจัดสมาชิกใหม่

ผลกับ throughput มาจากทางอ้อม การประมวลผลยังเดินได้เพราะการอ่านไม่ถูกปิด แต่งานที่ทำไปในช่วงนั้นยังไม่ถูกบันทึกว่าทำแล้ว ถ้าล้มระหว่างนั้นมันจะถูกทำซ้ำทั้งช่วง

คอร์สนี้ไม่มีตัวเลขของการ rebalance ให้เลย ไม่มีเวลาที่วัดได้ ไม่มีตัวคูณของ protocol รุ่นใหม่ และไม่มีอะไรใน expected/ ที่จะเอามารองรับตัวเลขแบบนั้นได้

ตัวคูณที่เจอบ่อยตามบทความของผู้ขายไม่มีที่มาแบบปฐมภูมิ คอร์สนี้จึงไม่พิมพ์มัน ข้ออ้างที่วัดไม่ได้ ไม่ควรมีตัวเลขติดมาด้วย

kaen-log ไม่ได้ทำอะไรบ้างในบทนี้

หัวข้อนี้ประกาศเส้นขอบเขตของ coordinator ที่บทนี้เพิ่งสร้าง

ไม่มี protocol ของกลุ่มบนสาย ไม่มี JoinGroup ไม่มี SyncGroup ไม่มี heartbeat จริง และไม่มีการเลือกผู้นำของกลุ่ม สมาชิกเข้ากลุ่มด้วยการเรียก method ในกระบวนการเดียวกัน
สถานะของกลุ่มอยู่ในหน่วยความจำล้วน Group ไม่เคยถูกเขียนลง disk ของ Kafka เก็บ metadata ของกลุ่มไว้ใน topic เดียวกับ offset แต่คอร์สนี้ไม่ได้ตรวจรูปของมัน จึงไม่อธิบายว่ามันหน้าตาอย่างไร
ไม่มีการหมดอายุของ offset ที่ commit ไว้ ของ Kafka มี และนาฬิกาเริ่มเดินตอนที่กลุ่มว่างลง ตาม GroupCoordinatorConfig ส่วน __offsets ของ kaen-log เก็บค่าไว้จนกว่าจะมีคนลบด้วย tombstone
coordinator ไม่ได้ถูกเลือกจากใคร ของ Kafka หา coordinator จากการ hash ชื่อกลุ่ม แล้ว mod ด้วยจำนวน partition ของ topic นั้น ตามที่ partitionFor เขียนไว้ว่า Utils.abs(groupId.hashCode()) % numPartitions การล้มของ coordinator จึงเป็นเรื่องเดียวกับ การล้มของ leader ของ partition ซึ่งเป็นกลไกที่บทที่ 8 ถึงจะมี
__offsets ไม่มีสำเนา ในขณะที่ค่าปริยายของ Kafka คือ replication factor 3

ข้อสุดท้ายมีผลกับข้ออ้างของทั้งบท offset ที่ commit แล้วของ kaen-log รอดจากการที่ process ตาย มันไม่ได้รอดจากการที่ disk ตาย และไม่มีบทไหนก่อนบทที่ 8 ที่อ้างแบบนั้นได้

บทที่ 6 · retention กับ compaction จะรับ __offsets ก้อนนี้ไปทำความสะอาด และตรงนั้นคือจุดที่ tombstone กับ delete horizon จะได้ทำงานจริงกับ log ที่บทนี้เพิ่งสร้าง

🔗 อ้างอิงต้นทางของบทนี้

KafkaConsumer.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — นิยามของตำแหน่งกับ committed position ประโยคที่ว่าความต่างของสองค่านี้ ให้ผู้อ่านคุมได้เองว่าเมื่อไหร่ถือว่า record ถูกบริโภค ประโยคที่บอกว่า offset ไม่รับประกันว่าติดกัน และประโยคที่บอกว่าตำแหน่งที่ตกลงไปในช่องว่างจะได้ record ตัวถัดไปที่ offset สูงกว่า ประโยคตัวหนาเรื่อง record ตัวถัดไป และคำเตือนเรื่อง auto commit ที่ทำให้ record หายได้
design.html tag 4.0.0 (ตรวจแล้ว 2026-09-03) — สองย่อหน้าที่นิยาม at-most-once กับ at-least-once ด้วยลำดับของขั้นตอนล้วน ประโยคหนึ่ง partition ต่อหนึ่งผู้อ่านในหนึ่งกลุ่ม และย่อหน้าที่บอกว่าคิวต้องเก็บสถานะหลายชั้น ต่อหนึ่งข้อความ
ClassicGroupState.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — รายการของแต่ละสถานะ ทั้งบรรทัดที่ปฏิเสธ commit และบรรทัด “allow offset fetch requests” ที่ปรากฏในทุกสถานะ
OffsetCommitKey.json tag 4.0.0 (ตรวจแล้ว 2026-09-03) — สาม field ของ key คือ group topic และ partition กับ validVersions ที่เป็น "0-1"
OffsetCommitRequest.json tag 4.0.0 (ตรวจแล้ว 2026-09-03) — field GenerationIdOrMemberEpoch ที่เป็น int32 มีค่าปริยาย -1 พร้อมคำอธิบายของมัน
GroupCoordinatorService.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — groupMetadataTopicConfigs ที่ใส่ CLEANUP_POLICY_COMPACT ให้ topic ของ offset และ partitionFor ที่ hash ชื่อกลุ่มแล้ว mod ด้วยจำนวน partition
GroupCoordinatorConfig.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — ค่าปริยาย 50 partition กับ replication factor 3 ของ topic นั้น คำอธิบายที่ว่าไม่ควรเปลี่ยนจำนวน partition หลัง deploy และการที่ offset ที่ commit ไว้มีวันหมดอายุ
ConsumerConfig.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — enable.auto.commit ที่ประกาศค่าปริยายเป็น true

เช็กความเข้าใจ — บทที่ 5

ข้อ 1 / 3

ผู้อ่านตัวหนึ่งล้มลง ตำแหน่งในหน่วยความจำอยู่ที่ 12 ส่วนค่าที่ commit ไว้บน disk เป็น 10 ผู้อ่านตัวใหม่ขึ้นมาแทน จะเกิดอะไรกับ order สองตัวที่อยู่ระหว่างสองเลขนั้น

เลข​สอง​ตัว​ของ​ผู้​บริโภค · ตำแหน่ง กับ offset ที่ commit แล้ว

ผู้​อ่าน​หนึ่ง​ราย​ถือ​เลข​สอง​ตัว และ​มี​ตัว​เดียว​ที่​รอด​จาก​การ​ล้ม

offset ที่ commit คือ record ตัว​ถัด​ไป ไม่ใช่​ตัว​สุดท้าย​ที่​ทำ​เสร็จ

หน้าต่าง​ของ​การ​ส่ง​ซ้ำ คือ​ผล​ลบ​ของ​เลข​สอง​ตัว

สลับ​สอง​บรรทัด แล้ว order หาย​จริง ไม่ใช่​แค่​ทำ​ซ้ำ

offset ที่ commit ไป​นั่ง​อยู่​ใน log ของ broker เอง

key ของ record คือ triple พอดี

commit หนึ่ง​ครั้ง​คือ produce หนึ่ง​ครั้ง

replay ของ __offsets และ tombstone ที่​บท​ที่ 6 จะ​เก็บ​กวาด

สาม​กลุ่ม อ่าน log เดียวกัน อยู่​คนละ​ตำแหน่ง

หนึ่ง partition ต่อ​หนึ่ง​สมาชิก​ใน​กลุ่ม​เดียวกัน

กัน​ผี​ด้วย​จำนวนเต็ม​ตัว​เดียว

ราคา​ที่​การ rebalance เก็บ​จาก​ทั้ง​กลุ่ม

kaen-log ไม่​ได้​ทำ​อะไร​บ้าง​ใน​บท​นี้