krabka_metadata/
directory_id.rs

1//! KIP-858 reserved log-directory ids, transcribed from Kafka's
2//! `org.apache.kafka.common.DirectoryId`.
3//!
4//! A partition's `directories` list holds one id per replica. Besides the real
5//! ids a broker generates for its log directories, Kafka reserves the first
6//! 100 UUIDs (most significant bits 0, least significant bits below 100) and
7//! gives three of them a meaning.
8
9use uuid::Uuid;
10
11/// `DirectoryId.MIGRATING`, `Uuid(0, 0)`: the replica already lives in some
12/// log directory, but nobody recorded which.
13///
14/// Kafka reads a partition record with no directories (every record before
15/// `3.7-IV2`) as all `MIGRATING`, and its controller places a new replica on a
16/// broker that registered no log-dir ids, or on any broker below `3.7-IV2`, as
17/// `MIGRATING`.
18pub const MIGRATING: Uuid = Uuid::from_u64_pair(0, 0);
19
20/// `DirectoryId.UNASSIGNED`, `Uuid(0, 1)`: a new replica that no log directory
21/// hosts yet.
22///
23/// Kafka's controller places a replica on a broker with more than one log
24/// directory, or on a broker it has no registration for, as `UNASSIGNED`, and
25/// the broker's `AssignReplicasToDirs` later names the directory.
26pub const UNASSIGNED: Uuid = Uuid::from_u64_pair(0, 1);
27
28/// `DirectoryId.LOST`, `Uuid(0, 2)`: the replica's directory is offline and
29/// its id is unknown.
30pub const LOST: Uuid = Uuid::from_u64_pair(0, 2);
31
32/// `DirectoryId.reserved`: true for the first 100 UUIDs, which no broker
33/// generates as a log-directory id.
34#[must_use]
35pub fn reserved(dir: Uuid) -> bool {
36    let (most, least) = dir.as_u64_pair();
37    most == 0 && least < 100
38}
39
40/// `DirectoryId.isOnline`: whether a replica assigned to `dir` counts as on
41/// an online directory of a broker whose online directories are `online_dirs`.
42///
43/// `UNASSIGNED` and `MIGRATING` are online and `LOST` is offline. An empty
44/// `online_dirs` also counts as online: Kafka reaches it only for a broker
45/// that registered before `3.7-IV2` carried log dirs, and it assumes a broker
46/// halts once every log directory is down.
47#[must_use]
48pub fn is_online(dir: Uuid, online_dirs: &[Uuid]) -> bool {
49    if dir == UNASSIGNED || dir == MIGRATING {
50        return true;
51    }
52    if dir == LOST {
53        return false;
54    }
55    online_dirs.is_empty() || online_dirs.contains(&dir)
56}
57
58#[cfg(test)]
59mod tests {
60    use assert2::check;
61
62    use super::*;
63
64    /// The sentinels are Kafka's `new Uuid(0L, 0L)`, `new Uuid(0L, 1L)` and
65    /// `new Uuid(0L, 2L)`, whose big-endian bytes end in 0, 1 and 2.
66    #[test]
67    fn sentinels_match_kafka() {
68        check!(
69            [MIGRATING, UNASSIGNED, LOST] == [Uuid::nil(), Uuid::from_u128(1), Uuid::from_u128(2)]
70        );
71    }
72
73    #[test]
74    fn reserved_is_the_first_hundred_uuids() {
75        for (case, dir, want) in [
76            ("MIGRATING", MIGRATING, true),
77            ("UNASSIGNED", UNASSIGNED, true),
78            ("LOST", LOST, true),
79            ("last reserved", Uuid::from_u128(99), true),
80            ("first unreserved", Uuid::from_u128(100), false),
81            ("high bits set", Uuid::from_u64_pair(1, 0), false),
82        ] {
83            check!(reserved(dir) == want, "{case}");
84        }
85    }
86
87    #[test]
88    fn is_online_matches_kafka() {
89        let (a, b) = (Uuid::from_u128(0xA00), Uuid::from_u128(0xB00));
90        for (case, dir, online, want) in [
91            ("UNASSIGNED", UNASSIGNED, &[a][..], true),
92            ("MIGRATING", MIGRATING, &[a][..], true),
93            ("LOST", LOST, &[a][..], false),
94            ("LOST with no dirs", LOST, &[][..], false),
95            ("listed dir", a, &[a, b][..], true),
96            ("unlisted dir", b, &[a][..], false),
97            ("no dirs registered", b, &[][..], true),
98        ] {
99            check!(is_online(dir, online) == want, "{case}");
100        }
101    }
102}
krabka_metadata/directory_id.rs

krabka_metadata/
directory_id.rs
