krabka_ids/
kafka_uuid.rs

1//! Kafka's 128-bit identifier and its base64url string form.
2
3use core::{cmp::Ordering, fmt, str::FromStr};
4
5use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
6use derive_more::{From, Into};
7use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
8use thiserror::Error;
9use uuid::Uuid;
10
11/// The longest input, in UTF-16 code units, that Kafka tries to decode.
12const MAX_ENCODED_LEN: usize = 24;
13
14/// The number of bytes in a UUID.
15const UUID_LEN: usize = 16;
16
17/// A Kafka `Uuid`: a cluster id, topic id, or log-directory id.
18///
19/// This is the value type of `org.apache.kafka.common.Uuid`. Kafka prints
20/// these ids as 22 characters of unpadded base64url, for example
21/// `UXyU9i5ARn6W00ON2taeWA`, not as the hyphenated hex form that
22/// [`uuid::Uuid`] prints. [`Display`](fmt::Display) and [`FromStr`] use the
23/// Kafka form. So does serde, which always writes the id as that string.
24///
25/// [`FromStr`] matches `Uuid.fromString` in Apache Kafka 4.3.1. It accepts
26/// exactly the inputs that Kafka accepts and gives the same error messages.
27/// Two consequences are not obvious:
28///
29/// - Kafka ignores the unused low bits of the last character, so
30///   `_____________________x` parses to the same id as
31///   `_____________________w`.
32/// - Kafka also accepts the 22 characters followed by `==`.
33///
34/// Parsing never rejects a reserved id. Kafka's `fromString` does not reject
35/// them, and `kafka-storage format --cluster-id` stores any string. Check
36/// [`is_reserved`](Self::is_reserved) where a caller must refuse one.
37///
38/// The ordering is Kafka's `compareTo`: it compares the two halves as signed
39/// 64-bit integers, most significant half first. This is not the unsigned
40/// byte order of [`uuid::Uuid`].
41///
42/// # Examples
43///
44/// ```
45/// use krabka_ids::KafkaUuid;
46///
47/// let id: KafkaUuid = "UXyU9i5ARn6W00ON2taeWA".parse()?;
48/// assert2::assert!(id.to_string() == "UXyU9i5ARn6W00ON2taeWA");
49/// assert2::assert!(KafkaUuid::METADATA_TOPIC_ID.to_string() == "AAAAAAAAAAAAAAAAAAAAAQ");
50/// # Ok::<(), krabka_ids::KafkaUuidError>(())
51/// ```
52#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, From, Into)]
53pub struct KafkaUuid(pub Uuid);
54
55impl KafkaUuid {
56    /// The id that stands for a null or empty id (Kafka's `ZERO_UUID`).
57    pub const ZERO: Self = Self::from_bits(0, 0);
58
59    /// The reserved id with value one (Kafka's `ONE_UUID`).
60    pub const ONE: Self = Self::from_bits(0, 1);
61
62    /// The topic id of the metadata topic, `__cluster_metadata`.
63    pub const METADATA_TOPIC_ID: Self = Self::ONE;
64
65    /// The ids that [`random`](Self::random) never returns (Kafka's
66    /// `Uuid.RESERVED`).
67    pub const RESERVED: [Self; 2] = [Self::ZERO, Self::ONE];
68
69    /// The directory id of a replica that is in a log directory that the
70    /// broker has not yet reported (Kafka's `DirectoryId.MIGRATING`).
71    pub const DIRECTORY_MIGRATING: Self = Self::ZERO;
72
73    /// The directory id of a replica that is not yet placed in a log directory
74    /// (Kafka's `DirectoryId.UNASSIGNED`).
75    pub const DIRECTORY_UNASSIGNED: Self = Self::ONE;
76
77    /// The directory id of a replica in an offline log directory that has no
78    /// known id (Kafka's `DirectoryId.LOST`).
79    pub const DIRECTORY_LOST: Self = Self::from_bits(0, 2);
80
81    /// Builds an id from its two halves, as Kafka's `Uuid(long, long)` does.
82    #[must_use]
83    pub const fn from_bits(most_significant: i64, least_significant: i64) -> Self {
84        Self(Uuid::from_u64_pair(
85            most_significant.cast_unsigned(),
86            least_significant.cast_unsigned(),
87        ))
88    }
89
90    /// The inner [`uuid::Uuid`].
91    #[must_use]
92    pub const fn get(self) -> Uuid {
93        self.0
94    }
95
96    /// The most significant 64 bits, as Kafka's `getMostSignificantBits`
97    /// returns them.
98    #[must_use]
99    pub const fn most_significant_bits(self) -> i64 {
100        self.0.as_u64_pair().0.cast_signed()
101    }
102
103    /// The least significant 64 bits, as Kafka's `getLeastSignificantBits`
104    /// returns them.
105    #[must_use]
106    pub const fn least_significant_bits(self) -> i64 {
107        self.0.as_u64_pair().1.cast_signed()
108    }
109
110    /// Returns a random version 4 id, as Kafka's `Uuid.randomUuid` does.
111    ///
112    /// The result is never in [`RESERVED`](Self::RESERVED), and its string
113    /// form never starts with `-`. A leading `-` would make a command-line
114    /// parser read the id as an option.
115    #[must_use]
116    pub fn random() -> Self {
117        first_accepted(Uuid::new_v4, Self::is_rejected_random)
118    }
119
120    /// Returns a random log-directory id, as Kafka's `DirectoryId.random`
121    /// does.
122    ///
123    /// The result obeys the rules of [`random`](Self::random), and it is never
124    /// one of the 100 reserved directory ids (see
125    /// [`is_reserved_directory_id`](Self::is_reserved_directory_id)).
126    #[must_use]
127    pub fn random_directory_id() -> Self {
128        first_accepted(Uuid::new_v4, Self::is_rejected_directory_id)
129    }
130
131    /// Returns `true` for an id in [`RESERVED`](Self::RESERVED).
132    #[must_use]
133    pub fn is_reserved(self) -> bool {
134        Self::RESERVED.contains(&self)
135    }
136
137    /// Returns `true` for one of the 100 lowest ids, which Kafka reserves as
138    /// directory ids (Kafka's `DirectoryId.reserved`).
139    #[must_use]
140    pub const fn is_reserved_directory_id(self) -> bool {
141        self.most_significant_bits() == 0 && self.least_significant_bits() < 100
142    }
143
144    /// Parses the Kafka form, or the 36-character hyphenated hex form.
145    ///
146    /// Kafka's `Uuid.fromString` rejects the hyphenated form, so
147    /// [`FromStr`] does too. Krabka's own tools also accept it, and they
148    /// parse through this function. An input with a `-` at each of the four
149    /// hyphen positions of a 36-character string is parsed as hex. Any other
150    /// input is parsed as [`FromStr`] parses it.
151    ///
152    /// # Errors
153    ///
154    /// Returns [`KafkaUuidError::Hyphenated`] when the input has the shape of
155    /// the hyphenated form but is not valid hex. Otherwise returns the error
156    /// of [`FromStr`].
157    pub fn parse_kafka_or_hyphenated(input: &str) -> Result<Self, KafkaUuidError> {
158        if has_hyphenated_shape(input) {
159            return Uuid::try_parse(input)
160                .map(Self)
161                .map_err(|source| KafkaUuidError::Hyphenated {
162                    input: input.to_owned(),
163                    source,
164                });
165        }
166        input.parse()
167    }
168
169    fn is_rejected_random(self) -> bool {
170        self.is_reserved() || self.string_starts_with_dash()
171    }
172
173    fn is_rejected_directory_id(self) -> bool {
174        self.is_rejected_random() || self.is_reserved_directory_id()
175    }
176
177    /// Returns `true` when the first base64url character is `-`, the
178    /// character for the six-bit value 62.
179    fn string_starts_with_dash(self) -> bool {
180        self.0.as_bytes()[0] >> 2 == 62
181    }
182}
183
184/// Draws ids from `next` until one is not rejected.
185fn first_accepted(mut next: impl FnMut() -> Uuid, rejected: fn(KafkaUuid) -> bool) -> KafkaUuid {
186    loop {
187        let candidate = KafkaUuid(next());
188        if !rejected(candidate) {
189            return candidate;
190        }
191    }
192}
193
194fn has_hyphenated_shape(input: &str) -> bool {
195    let bytes = input.as_bytes();
196    bytes.len() == 36 && [8, 13, 18, 23].iter().all(|&i| bytes.get(i) == Some(&b'-'))
197}
198
199impl fmt::Display for KafkaUuid {
200    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
201        f.pad(&URL_SAFE_NO_PAD.encode(self.0.as_bytes()))
202    }
203}
204
205impl FromStr for KafkaUuid {
206    type Err = KafkaUuidError;
207
208    /// Parses the Kafka form, as Kafka's `Uuid.fromString` does.
209    fn from_str(input: &str) -> Result<Self, Self::Err> {
210        // Kafka measures the length of a Java string, in UTF-16 code units.
211        if input.encode_utf16().count() > MAX_ENCODED_LEN {
212            return Err(KafkaUuidError::TooLong {
213                prefix: java_prefix(input, MAX_ENCODED_LEN),
214            });
215        }
216        let decoded = java_url_decode(&latin1_bytes(input))?;
217        let bytes: [u8; UUID_LEN] =
218            decoded
219                .try_into()
220                .map_err(|decoded: Vec<u8>| KafkaUuidError::WrongLength {
221                    input: input.to_owned(),
222                    decoded_len: decoded.len(),
223                })?;
224        Ok(Self(Uuid::from_bytes(bytes)))
225    }
226}
227
228impl PartialOrd for KafkaUuid {
229    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
230        Some(self.cmp(other))
231    }
232}
233
234impl Ord for KafkaUuid {
235    fn cmp(&self, other: &Self) -> Ordering {
236        (self.most_significant_bits(), self.least_significant_bits()).cmp(&(
237            other.most_significant_bits(),
238            other.least_significant_bits(),
239        ))
240    }
241}
242
243impl Serialize for KafkaUuid {
244    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
245        serializer.collect_str(self)
246    }
247}
248
249impl<'de> Deserialize<'de> for KafkaUuid {
250    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
251        String::deserialize(deserializer)?
252            .parse()
253            .map_err(D::Error::custom)
254    }
255}
256
257/// A string that is not a valid [`KafkaUuid`].
258///
259/// Each message is the message of the `IllegalArgumentException` that
260/// Kafka's `Uuid.fromString` throws for the same input. Kafka passes the
261/// messages of `java.util.Base64` through unchanged, so some of them name a
262/// byte array.
263#[derive(Debug, Clone, PartialEq, Eq, Error)]
264#[non_exhaustive]
265pub enum KafkaUuidError {
266    /// The input is longer than 24 UTF-16 code units.
267    #[error("Input string with prefix `{prefix}` is too long to be decoded as a base64 UUID")]
268    TooLong {
269        /// The first 24 UTF-16 code units of the input.
270        prefix: String,
271    },
272
273    /// The input is valid base64url, but not of 16 bytes.
274    #[error(
275        "Input string `{input}` decoded as {decoded_len} bytes, which is not equal to the \
276         expected 16 bytes of a base64-encoded UUID"
277    )]
278    WrongLength {
279        /// The input.
280        input: String,
281        /// The number of bytes that the input decodes to.
282        decoded_len: usize,
283    },
284
285    /// The input is a single character.
286    #[error("Input byte[] should at least have 2 bytes for base64 bytes")]
287    TooShort,
288
289    /// The input holds a byte that is not in the base64url alphabet.
290    #[error("Illegal base64 character {}", JavaSignedHex(*.byte))]
291    IllegalCharacter {
292        /// The byte. A character above U+00FF is the byte `?`.
293        byte: u8,
294    },
295
296    /// A `=` is in a position where base64 does not allow padding.
297    #[error("Input byte array has wrong 4-byte ending unit")]
298    WrongEndingUnit,
299
300    /// The last base64 unit has only one character.
301    #[error("Last unit does not have enough valid bits")]
302    LastUnitTooShort,
303
304    /// Bytes follow the padding.
305    #[error("Input byte array has incorrect ending byte at {position}")]
306    IncorrectEndingByte {
307        /// The zero-based index of the first byte after the padding.
308        position: usize,
309    },
310
311    /// The input has the shape of the hyphenated hex form but is not valid
312    /// hex. Only [`KafkaUuid::parse_kafka_or_hyphenated`] returns it.
313    #[error("Input string `{input}` is not a valid hyphenated UUID: {source}")]
314    Hyphenated {
315        /// The input.
316        input: String,
317        /// The error from the hex parser.
318        source: uuid::Error,
319    },
320}
321
322/// Formats a byte as Java's `Integer.toString(b, 16)` formats a signed
323/// `byte`: `0xe9` is `-17`.
324struct JavaSignedHex(u8);
325
326impl fmt::Display for JavaSignedHex {
327    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
328        let value = self.0.cast_signed();
329        if value < 0 {
330            write!(f, "-{:x}", value.unsigned_abs())
331        } else {
332            write!(f, "{value:x}")
333        }
334    }
335}
336
337/// Returns the first `max` UTF-16 code units of `input`, as Java's
338/// `substring(0, max)` does.
339///
340/// When the cut splits a surrogate pair, Java keeps the lone high surrogate,
341/// and the JVM prints it as `?`. This function writes that `?`.
342fn java_prefix(input: &str, max: usize) -> String {
343    let mut prefix = String::new();
344    let mut units = 0;
345    for c in input.chars() {
346        let width = c.len_utf16();
347        if units + width > max {
348            if units < max {
349                prefix.push('?');
350            }
351            break;
352        }
353        prefix.push(c);
354        units += width;
355    }
356    prefix
357}
358
359/// Encodes `input` as Java's `String.getBytes(ISO_8859_1)` does: one byte per
360/// character, and `?` for a character above U+00FF.
361fn latin1_bytes(input: &str) -> Vec<u8> {
362    input
363        .chars()
364        .map(|c| u8::try_from(c).unwrap_or(b'?'))
365        .collect()
366}
367
368/// The six-bit value of a base64url character.
369fn url_value(byte: u8) -> Option<u32> {
370    let value = match byte {
371        b'A'..=b'Z' => byte - b'A',
372        b'a'..=b'z' => byte - b'a' + 26,
373        b'0'..=b'9' => byte - b'0' + 52,
374        b'-' => 62,
375        b'_' => 63,
376        _ => return None,
377    };
378    Some(u32::from(value))
379}
380
381/// Decodes base64url as `Base64.getUrlDecoder().decode(byte[])` does in the
382/// JDK that Kafka 4.3.1 ships with.
383///
384/// This decoder exists so that [`KafkaUuid`] accepts and rejects the same
385/// strings as Kafka, with the same messages. The `base64` crate is stricter:
386/// it rejects the non-zero trailing bits that the JDK ignores. The caller
387/// bounds the input to 24 bytes, so the output is at most 18 bytes.
388fn java_url_decode(src: &[u8]) -> Result<Vec<u8>, KafkaUuidError> {
389    // `decodedOutLength`: the only check that it makes before decoding.
390    if src.len() == 1 {
391        return Err(KafkaUuidError::TooShort);
392    }
393
394    // `decode0`, without the MIME branches.
395    let mut dst = Vec::with_capacity(src.len() / 4 * 3 + 2);
396    let mut bits = 0u32;
397    let mut shift_to = 18i32;
398    let mut sp = 0;
399    while let Some(&byte) = src.get(sp) {
400        sp += 1;
401        if let Some(value) = url_value(byte) {
402            bits |= value << shift_to;
403            shift_to -= 6;
404            if shift_to < 0 {
405                dst.extend_from_slice(&bits.to_be_bytes()[1..]);
406                shift_to = 18;
407                bits = 0;
408            }
409            continue;
410        }
411        if byte != b'=' {
412            return Err(KafkaUuidError::IllegalCharacter { byte });
413        }
414        // `=` ends the data. Only `xx==` and `xxx=` are complete units.
415        let second_pad_missing = shift_to == 6
416            && src.get(sp).is_none_or(|&next| {
417                sp += 1;
418                next != b'='
419            });
420        if second_pad_missing || shift_to == 18 {
421            return Err(KafkaUuidError::WrongEndingUnit);
422        }
423        break;
424    }
425    match shift_to {
426        6 => dst.push(bits.to_be_bytes()[1]),
427        0 => dst.extend_from_slice(&bits.to_be_bytes()[1..3]),
428        12 => return Err(KafkaUuidError::LastUnitTooShort),
429        _ => {}
430    }
431    if sp < src.len() {
432        return Err(KafkaUuidError::IncorrectEndingByte { position: sp });
433    }
434    Ok(dst)
435}
436
437#[cfg(test)]
438mod tests {
439    use std::collections::VecDeque;
440
441    use uuid::Uuid;
442
443    use super::{KafkaUuid, first_accepted};
444
445    type Rejected = fn(KafkaUuid) -> bool;
446
447    // The first id that the two generators accept, from a fixed sequence of
448    // candidates. `-` leads the string form of any id whose first byte is in
449    // `0xf8..=0xfb`.
450    #[test]
451    fn random_generators_skip_the_ids_that_kafka_skips() {
452        let dash = Uuid::from_u128(0xf800_0000_0000_4000_8000_0000_0000_0000);
453        let dash_last = Uuid::from_u128(0xfbff_ffff_ffff_4fff_bfff_ffff_ffff_ffff);
454        let not_dash = Uuid::from_u128(0xfc00_0000_0000_4000_8000_0000_0000_0000);
455        let reserved_dir = Uuid::from_u128(99);
456        let first_free_dir = Uuid::from_u128(100);
457
458        let cases: [(&str, Vec<Uuid>, Rejected, Uuid); 5] = [
459            (
460                "random skips ZERO and ONE",
461                vec![Uuid::nil(), Uuid::from_u128(1), reserved_dir],
462                KafkaUuid::is_rejected_random,
463                reserved_dir,
464            ),
465            (
466                "random skips a leading dash",
467                vec![dash, dash_last, not_dash],
468                KafkaUuid::is_rejected_random,
469                not_dash,
470            ),
471            (
472                "directory id skips the random rejects",
473                vec![Uuid::nil(), dash, first_free_dir],
474                KafkaUuid::is_rejected_directory_id,
475                first_free_dir,
476            ),
477            (
478                "directory id skips the first 100",
479                vec![Uuid::from_u128(2), reserved_dir, first_free_dir],
480                KafkaUuid::is_rejected_directory_id,
481                first_free_dir,
482            ),
483            (
484                "directory id takes the msb-nonzero id",
485                vec![Uuid::from_u64_pair(1, 0)],
486                KafkaUuid::is_rejected_directory_id,
487                Uuid::from_u64_pair(1, 0),
488            ),
489        ];
490        for (name, candidates, rejected, expected) in cases {
491            let mut queue = VecDeque::from(candidates);
492            let chosen = first_accepted(|| queue.pop_front().expect("ran out"), rejected);
493            assert2::assert!(chosen == KafkaUuid(expected), "{name}");
494        }
495    }
496
497    #[test]
498    fn the_dash_check_agrees_with_the_string_form() {
499        for first_byte in 0..=u8::MAX {
500            let mut bytes = [0u8; 16];
501            bytes[0] = first_byte;
502            let id = KafkaUuid(Uuid::from_bytes(bytes));
503            assert2::assert!(
504                id.string_starts_with_dash() == id.to_string().starts_with('-'),
505                "{first_byte:#x}"
506            );
507        }
508    }
509}
krabka_ids/kafka_uuid.rs

krabka_ids/
kafka_uuid.rs
