krabka_format/
ids.rs

1//! The two identity UUIDs that `format` writes, in Kafka's string form.
2//!
3//! `format` makes or accepts a **cluster id**, and makes a **directory id** for
4//! each log directory. The directory id of the metadata directory is also the
5//! node's KIP-853 voter identity. Both ids go into `meta.properties`, and the
6//! cluster id also goes into the bootstrap manifest.
7//!
8//! Kafka prints an `org.apache.kafka.common.Uuid` as 22 characters of unpadded
9//! base64url, for example `AQIDBAUGBwgJCgsMDQ4PEA`, not as the hyphenated hex
10//! form that [`uuid::Uuid`] prints. [`ClusterId`] and [`DirectoryId`] use the
11//! Kafka form for [`Display`](core::fmt::Display), [`FromStr`](core::str::FromStr),
12//! and serde. An id that `format` writes can thus go into a Kafka tool, and an
13//! id from a Kafka cluster can go into `format`.
14//!
15//! [`FromStr`](core::str::FromStr) accepts the same strings as Kafka 4.3.1's
16//! `Uuid.fromString`:
17//! 22 base64url characters, optionally followed by `==`, with the unused low
18//! bits of the last character ignored. Parsing does not refuse the two
19//! reserved ids, because Kafka does not: `kafka-storage format -t
20//! AAAAAAAAAAAAAAAAAAAAAA` succeeds on `apache/kafka:4.3.1`. The command line
21//! also accepts the 36-character hyphenated form, through
22//! [`ClusterId::parse_cli`] and [`DirectoryId::parse_cli`]. A file is read in
23//! the Kafka form only.
24//!
25//! Both ids wrap a `uuid::Uuid`, so a bare-`Uuid` signature such as
26//! `write_meta_properties(_, cluster, dir)` would compile with the two
27//! arguments transposed. A distinct newtype for each id makes the compiler
28//! reject that mix-up.
29
30use {
31    base64::{
32        Engine as _,
33        alphabet::URL_SAFE,
34        engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig},
35    },
36    derive_more::{From, Into},
37    thiserror::Error,
38    uuid::Uuid,
39};
40
41/// The longest input, in UTF-16 code units, that Kafka tries to decode.
42const MAX_ENCODED_LEN: usize = 24;
43
44/// Unpadded base64url that, as the JDK decoder does, ignores the unused low
45/// bits of the last character.
46const KAFKA_BASE64: GeneralPurpose = GeneralPurpose::new(
47    &URL_SAFE,
48    GeneralPurposeConfig::new()
49        .with_encode_padding(false)
50        .with_decode_allow_trailing_bits(true)
51        .with_decode_padding_mode(DecodePaddingMode::RequireNone),
52);
53
54/// A Kafka cluster id: the identity that every node of a cluster shares.
55///
56/// This id is distinct from [`DirectoryId`], which is per log directory.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, From, Into, krabka_macros::KafkaUuid)]
58pub struct ClusterId(pub Uuid);
59
60/// A log directory's stable id (KIP-858). The metadata directory's id is also
61/// the node's KIP-853 voter identity.
62///
63/// `format` writes this id to `meta.properties`, and the broker reads it back
64/// on every boot. The type is distinct from [`ClusterId`], so a call site
65/// cannot transpose the two.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, From, Into, krabka_macros::KafkaUuid)]
67pub struct DirectoryId(pub Uuid);
68
69impl ClusterId {
70    /// Returns a random id, as Kafka's `Uuid.randomUuid` does.
71    ///
72    /// The result is never one of Kafka's two reserved ids, and its string
73    /// form never starts with `-`, which a command-line parser would read as
74    /// an option.
75    #[must_use]
76    pub fn random() -> Self { Self(random_uuid()) }
77
78    /// Parses a command-line value: the Kafka form, or the hyphenated form.
79    ///
80    /// # Errors
81    ///
82    /// Returns [`KafkaUuidError::Hyphenated`] for an input that has the shape
83    /// of the hyphenated form but is not hex. Otherwise returns the error of
84    /// [`FromStr`](core::str::FromStr).
85    pub fn parse_cli(input: &str) -> Result<Self, KafkaUuidError> {
86        parse_kafka_or_hyphenated(input).map(Self)
87    }
88}
89
90impl DirectoryId {
91    /// Returns a random directory id, as Kafka's `DirectoryId.random` does.
92    ///
93    /// The result obeys the rules of [`ClusterId::random`], and it is never
94    /// one of the 100 lowest ids, which Kafka reserves as directory-id
95    /// sentinels.
96    #[must_use]
97    pub fn random() -> Self {
98        Self(first_accepted(|id| {
99            is_reserved_directory_id(id) || starts_with_dash(id)
100        }))
101    }
102
103    /// Returns a random directory id that is not in `used`, as Kafka's
104    /// `MetaPropertiesEnsemble.Copier.generateValidDirectoryId` returns one
105    /// that no other directory of the node holds.
106    #[must_use]
107    pub fn random_unused(used: &[Self]) -> Self {
108        loop {
109            let id = Self::random();
110            if !used.contains(&id) {
111                return id;
112            }
113        }
114    }
115
116    /// Returns `true` for one of the 100 lowest ids, which Kafka reserves as
117    /// directory ids (Kafka's `DirectoryId.reserved`).
118    #[must_use]
119    pub fn is_reserved(self) -> bool { is_reserved_directory_id(self.0) }
120
121    /// Parses a command-line value: the Kafka form, or the hyphenated form.
122    ///
123    /// # Errors
124    ///
125    /// As [`ClusterId::parse_cli`].
126    pub fn parse_cli(input: &str) -> Result<Self, KafkaUuidError> {
127        parse_kafka_or_hyphenated(input).map(Self)
128    }
129}
130
131/// Returns a random id, as Kafka's `Uuid.randomUuid` does: never the zero id
132/// or the id with value one, and never one whose string form starts with `-`,
133/// which a command-line parser would read as an option.
134///
135/// [`ClusterId::random`] uses it. So does the broker for a new topic id, which
136/// Kafka's `ReplicationControlManager.createTopic` draws the same way.
137#[must_use]
138pub fn random_uuid() -> Uuid { first_accepted(|id| is_reserved(id) || starts_with_dash(id)) }
139
140/// Kafka's `Uuid.RESERVED`: the zero id and the id with value one.
141fn is_reserved(id: Uuid) -> bool { id.as_u128() <= 1 }
142
143/// Kafka's `DirectoryId.reserved`: the most significant half is zero, and the
144/// least significant half, as a signed integer, is below 100.
145fn is_reserved_directory_id(id: Uuid) -> bool {
146    let (most, least) = id.as_u64_pair();
147    most == 0 && least.cast_signed() < 100
148}
149
150/// Returns `true` when the first base64url character is `-`, which stands for
151/// the six-bit value 62.
152fn starts_with_dash(id: Uuid) -> bool { id.as_bytes()[0] >> 2 == 62 }
153
154/// Draws version 4 ids until one is not rejected.
155fn first_accepted(rejected: impl Fn(Uuid) -> bool) -> Uuid {
156    loop {
157        let candidate = Uuid::new_v4();
158        if !rejected(candidate) {
159            return candidate;
160        }
161    }
162}
163
164/// Kafka's `Uuid.toString`: unpadded base64url of the 16 bytes.
165fn encode(id: Uuid) -> String { KAFKA_BASE64.encode(id.as_bytes()) }
166
167/// Kafka's `Uuid.fromString`.
168///
169/// Kafka accepts exactly 22 base64url characters, optionally followed by
170/// `==`. It measures the length of a Java string, in UTF-16 code units, and
171/// refuses more than 24 before it decodes anything.
172fn decode(input: &str) -> Result<Uuid, KafkaUuidError> {
173    if input.encode_utf16().count() > MAX_ENCODED_LEN {
174        return Err(KafkaUuidError::TooLong {
175            prefix: java_prefix(input, MAX_ENCODED_LEN),
176        });
177    }
178    let unpadded = input.strip_suffix("==").unwrap_or(input);
179    let bytes = KAFKA_BASE64
180        .decode(unpadded)
181        .map_err(|error| KafkaUuidError::NotBase64 {
182            input: input.to_owned(),
183            reason: error.to_string(),
184        })?;
185    let Ok(uuid) = <[u8; 16]>::try_from(bytes.as_slice()) else {
186        return Err(KafkaUuidError::WrongLength {
187            input: input.to_owned(),
188            decoded_len: bytes.len(),
189        });
190    };
191    Ok(Uuid::from_bytes(uuid))
192}
193
194/// Parses the hyphenated form when the input has its shape, and the Kafka
195/// form otherwise.
196fn parse_kafka_or_hyphenated(input: &str) -> Result<Uuid, KafkaUuidError> {
197    let bytes = input.as_bytes();
198    if bytes.len() == 36 && [8, 13, 18, 23].iter().all(|&i| bytes.get(i) == Some(&b'-')) {
199        return Uuid::try_parse(input).map_err(|error| KafkaUuidError::Hyphenated {
200            input: input.to_owned(),
201            reason: error.to_string(),
202        });
203    }
204    decode(input)
205}
206
207/// Returns the first `max` UTF-16 code units of `input`, as Java's
208/// `substring(0, max)` does. A cut that splits a surrogate pair keeps the lone
209/// high surrogate, which the JVM prints as `?`.
210fn java_prefix(input: &str, max: usize) -> String {
211    let mut prefix = String::new();
212    let mut units = 0;
213    for c in input.chars() {
214        let width = c.len_utf16();
215        if units + width > max {
216            if units < max {
217                prefix.push('?');
218            }
219            break;
220        }
221        prefix.push(c);
222        units += width;
223    }
224    prefix
225}
226
227/// A string that is not a Kafka `Uuid`.
228///
229/// [`TooLong`](Self::TooLong) and [`WrongLength`](Self::WrongLength) carry the
230/// message of the `IllegalArgumentException` that Kafka's `Uuid.fromString`
231/// throws for the same input. Kafka's other messages come from
232/// `java.util.Base64`; this type names the problem in its own words there.
233#[derive(Debug, Clone, PartialEq, Eq, Error)]
234#[non_exhaustive]
235pub enum KafkaUuidError {
236    /// The input is longer than 24 UTF-16 code units.
237    #[error("Input string with prefix `{prefix}` is too long to be decoded as a base64 UUID")]
238    TooLong {
239        /// The first 24 UTF-16 code units of the input.
240        prefix: String,
241    },
242
243    /// The input is base64url, but not of 16 bytes.
244    #[error(
245        "Input string `{input}` decoded as {decoded_len} bytes, which is not equal to the \
246         expected 16 bytes of a base64-encoded UUID"
247    )]
248    WrongLength {
249        /// The input.
250        input: String,
251        /// The number of bytes that the input decodes to.
252        decoded_len: usize,
253    },
254
255    /// The input is not base64url.
256    #[error("Input string `{input}` is not base64url: {reason}")]
257    NotBase64 {
258        /// The input.
259        input: String,
260        /// What the decoder refused.
261        reason: String,
262    },
263
264    /// The input has the shape of the hyphenated hex form but is not hex.
265    /// Only the `parse_cli` functions return it.
266    #[error("Input string `{input}` is not a valid hyphenated UUID: {reason}")]
267    Hyphenated {
268        /// The input.
269        input: String,
270        /// What the hex parser refused.
271        reason: String,
272    },
273}
274
275#[cfg(test)]
276mod tests {
277    use {super::*, assert2::check};
278
279    /// Every accepted input renders as the string that
280    /// `org.apache.kafka.common.Uuid` in `kafka-clients-4.3.1.jar` prints for
281    /// it, and every input Kafka refuses is refused.
282    #[yare::parameterized(
283        kafka_cluster = { "AQIDBAUGBwgJCgsMDQ4PEA", Ok("AQIDBAUGBwgJCgsMDQ4PEA") },
284        kafka_directory = { "U-hCdTzaSPyFB3swwc7Qdw", Ok("U-hCdTzaSPyFB3swwc7Qdw") },
285        reserved_zero = { "AAAAAAAAAAAAAAAAAAAAAA", Ok("AAAAAAAAAAAAAAAAAAAAAA") },
286        reserved_one = { "AAAAAAAAAAAAAAAAAAAAAQ", Ok("AAAAAAAAAAAAAAAAAAAAAQ") },
287        ignored_low_bits = { "_____________________x", Ok("_____________________w") },
288        rounded_low_bits = { "AAAAAAAAAAAAAAAAAAAAAB", Ok("AAAAAAAAAAAAAAAAAAAAAA") },
289        padded = { "AAAAAAAAAAAAAAAAAAAAAA==", Ok("AAAAAAAAAAAAAAAAAAAAAA") },
290        hyphenated_is_too_long = {
291            "01020304-0506-0708-090a-0b0c0d0e0f10",
292            Err(
293                "Input string with prefix `01020304-0506-0708-090a-` is too long to be decoded as a base64 UUID",
294            ),
295        },
296        unicode_too_long = {
297            "AAAAAAAAAAAAAAAAAAAAAAA😀",
298            Err(
299                "Input string with prefix `AAAAAAAAAAAAAAAAAAAAAAA?` is too long to be decoded as a base64 UUID",
300            ),
301        },
302        twenty_four_chars = {
303            "AAAAAAAAAAAAAAAAAAAAAAAA",
304            Err(
305                "Input string `AAAAAAAAAAAAAAAAAAAAAAAA` decoded as 18 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
306            ),
307        },
308        empty = {
309            "",
310            Err(
311                "Input string `` decoded as 0 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
312            ),
313        },
314        short = {
315            "AAAAAA",
316            Err(
317                "Input string `AAAAAA` decoded as 4 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
318            ),
319        },
320        bad_padding_length = {
321            "AAAAAAAAAAAAAAAAAAAA==",
322            Err(
323                "Input string `AAAAAAAAAAAAAAAAAAAA==` decoded as 15 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
324            ),
325        },
326        twenty_one_chars = {
327            "AAAAAAAAAAAAAAAAAAAAA",
328            Err("Input string `AAAAAAAAAAAAAAAAAAAAA` is not base64url: Invalid input length: 21"),
329        },
330        invalid_plus = {
331            "AAAAAAAAAAAAAAAAAAAA+A",
332            Err("Input string `AAAAAAAAAAAAAAAAAAAA+A` is not base64url: Invalid symbol 43, offset 20."),
333        },
334        bad_padding = {
335            "AAAAAAAAAAAAAAAAAAAAAA=",
336            Err("Input string `AAAAAAAAAAAAAAAAAAAAAA=` is not base64url: Invalid padding"),
337        },
338        non_ascii = {
339            "AAAAAAAAAAAAAAAAAAAAAé",
340            Err("Input string `AAAAAAAAAAAAAAAAAAAAAé` is not base64url: Invalid symbol 195, offset 21."),
341        },
342    )]
343    fn parse_and_render_match_kafka_uuid(input: &str, want: Result<&str, &str>) {
344        let want = want.map(str::to_owned).map_err(str::to_owned);
345        let as_cluster = input
346            .parse::<ClusterId>()
347            .map(|id| id.to_string())
348            .map_err(|e| e.to_string());
349        assert2::check!(as_cluster == want, "{input:?}");
350        let as_directory = input
351            .parse::<DirectoryId>()
352            .map(|id| id.to_string())
353            .map_err(|e| e.to_string());
354        assert2::check!(as_directory == want, "{input:?} as a directory id");
355    }
356
357    /// The command line also takes the hyphenated form, and only a string of
358    /// exactly that shape is read as hex.
359    #[yare::parameterized(
360        kafka_form = { "AQIDBAUGBwgJCgsMDQ4PEA" },
361        lowercase_hex = { "01020304-0506-0708-090a-0b0c0d0e0f10" },
362        uppercase_hex = { "01020304-0506-0708-090A-0B0C0D0E0F10" },
363    )]
364    fn the_command_line_accepts_both_forms(input: &str) {
365        let id = Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10);
366        assert2::check!(ClusterId::parse_cli(input) == Ok(ClusterId(id)), "{input}");
367        assert2::check!(
368            DirectoryId::parse_cli(input) == Ok(DirectoryId(id)),
369            "{input}"
370        );
371    }
372
373    #[test]
374    fn the_command_line_refuses_malformed_hex_and_other_uuid_spellings() {
375        let not_hex = "01020304-0506-0708-090a-0b0c0d0e0f1g";
376        check!(matches!(
377            ClusterId::parse_cli(not_hex),
378            Err(KafkaUuidError::Hyphenated { input, .. }) if input == not_hex
379        ));
380        // The other spellings `uuid` accepts are not the hyphenated form.
381        check!(matches!(
382            ClusterId::parse_cli("0102030405060708090a0b0c0d0e0f10"),
383            Err(KafkaUuidError::TooLong { .. })
384        ));
385    }
386
387    /// Serde carries the `Display` string and refuses the hyphenated form, so
388    /// a file holds only the Kafka form.
389    #[test]
390    fn serde_uses_the_kafka_string() {
391        let id = ClusterId(Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10));
392        check!(serde_json::to_string(&id).unwrap() == r#""AQIDBAUGBwgJCgsMDQ4PEA""#);
393        check!(serde_json::from_str::<ClusterId>(r#""AQIDBAUGBwgJCgsMDQ4PEA""#).unwrap() == id);
394        check!(
395            serde_json::from_str::<DirectoryId>(r#""01020304-0506-0708-090a-0b0c0d0e0f10""#)
396                .is_err()
397        );
398    }
399
400    /// The generators skip what Kafka's generators skip.
401    #[test]
402    fn random_ids_obey_kafka_generation_rules() {
403        // One draw in 64 starts with a dash before the rule applies, so
404        // 5000 draws would miss a missing rule with a chance of e^-78.
405        for _ in 0..5000 {
406            let plain = random_uuid();
407            check!(!is_reserved(plain));
408            check!(!encode(plain).starts_with('-'));
409        }
410        for _ in 0..500 {
411            let cluster = ClusterId::random();
412            check!(!is_reserved(cluster.0));
413            check!(!cluster.to_string().starts_with('-'));
414            let directory = DirectoryId::random();
415            check!(!directory.is_reserved());
416            check!(!directory.to_string().starts_with('-'));
417        }
418    }
419
420    /// The reservation and dash predicates agree with Kafka's definitions.
421    #[yare::parameterized(
422        zero = { Uuid::from_u64_pair(0, 0), true, true },
423        one = { Uuid::from_u64_pair(0, 1), true, true },
424        two = { Uuid::from_u64_pair(0, 2), false, true },
425        ninety_nine = { Uuid::from_u64_pair(0, 99), false, true },
426        one_hundred = { Uuid::from_u64_pair(0, 100), false, false },
427        low_bits_max = { Uuid::from_u64_pair(0, u64::MAX), false, true },
428        high_bits_one = { Uuid::from_u64_pair(1, 0), false, false },
429        high_and_low_one = { Uuid::from_u64_pair(1, 1), false, false },
430    )]
431    fn reservations_match_kafka(id: Uuid, reserved: bool, reserved_directory: bool) {
432        assert2::check!(
433            (is_reserved(id), is_reserved_directory_id(id)) == (reserved, reserved_directory),
434            "{id}"
435        );
436    }
437
438    /// Exhaustively checks each possible leading byte; this is a property
439    /// sweep over the byte domain, rather than a finite set of independent ids.
440    #[test]
441    fn dash_check_matches_kafka_for_every_first_byte() {
442        for first_byte in 0..=u8::MAX {
443            let mut bytes = [0u8; 16];
444            bytes[0] = first_byte;
445            let id = Uuid::from_bytes(bytes);
446            check!(
447                starts_with_dash(id) == encode(id).starts_with('-'),
448                "{first_byte:#x}"
449            );
450        }
451    }
452}
krabka_format/ids.rs

krabka_format/
ids.rs
