krabka_format/
meta_properties.rs

1//! Kafka's `meta.properties`: the file that marks a formatted log directory
2//! and records the identity of the node and of the directory.
3//!
4//! `kafka-storage format` writes the file into every directory it formats,
5//! and the broker reads it from every directory on every boot. Kafka's tools
6//! read it too: `kafka-metadata-quorum add-controller` takes the directory id
7//! of the new controller from the `meta.properties` in its
8//! `metadata.log.dir`. krabka writes and reads the same file, so a directory
9//! that one side formats is a formatted directory to the other side.
10//!
11//! The file is a Java properties file, written by `Properties.store` through
12//! Kafka's `PropertiesUtils.writePropertiesFile`. A `KRaft` node writes
13//! version 1 (`MetaPropertiesVersion.V1`):
14//!
15//! ```text
16//! #
17//! #Thu Feb 29 12:34:56 UTC 2024
18//! cluster.id=AQIDBAUGBwgJCgsMDQ4PEA
19//! directory.id=U-hCdTzaSPyFB3swwc7Qdw
20//! node.id=1
21//! version=1
22//! ```
23//!
24//! The first line is the empty comment that `writePropertiesFile` passes to
25//! `store`. The second line is the time of the write, in the form of
26//! `java.util.Date.toString`. The keys follow in their natural order, which
27//! is the order `Properties.store` uses on Java 18 and later. [`MetaProperties`]
28//! writes the same bytes, with the time in UTC. It reads what
29//! `Properties.load` reads, so it reads the file of every Java version.
30
31use {
32    crate::ids::{ClusterId, DirectoryId, KafkaUuidError},
33    std::{
34        collections::HashMap,
35        fs,
36        io::{self, Write as _},
37        mem,
38        path::Path,
39        time::{SystemTime, UNIX_EPOCH},
40    },
41    thiserror::Error,
42    time::OffsetDateTime,
43};
44
45/// The file name, Kafka's `MetaPropertiesEnsemble.META_PROPERTIES_NAME`.
46pub const META_PROPERTIES: &str = "meta.properties";
47
48/// The name of the file while it is written, before the rename that
49/// publishes it. `PropertiesUtils.writePropertiesFile` adds `.tmp`.
50pub const META_PROPERTIES_TMP: &str = "meta.properties.tmp";
51
52/// The version of the file that a `KRaft` node writes,
53/// `MetaPropertiesVersion.V1`.
54pub const META_PROPERTIES_VERSION: i32 = 1;
55
56/// The property names, from Kafka's `MetaProperties`.
57const VERSION_PROP: &str = "version";
58const CLUSTER_ID_PROP: &str = "cluster.id";
59const NODE_ID_PROP: &str = "node.id";
60const DIRECTORY_ID_PROP: &str = "directory.id";
61
62/// The content of a version 1 `meta.properties`.
63///
64/// Kafka's `MetaProperties` keeps the cluster id as a string. krabka keeps it
65/// as 16 bytes, so a cluster id that is not a Kafka `Uuid` does not read.
66/// The node id is Kafka's `int`. The directory id is optional when the file
67/// is read, as in Kafka: a broker gives a directory without one an id at its
68/// next start.
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70pub struct MetaProperties {
71    /// The cluster that the directory belongs to.
72    pub cluster_id: ClusterId,
73    /// The node that the directory belongs to, Kafka's `node.id`.
74    pub node_id: i32,
75    /// The directory's own id (KIP-858).
76    pub directory_id: Option<DirectoryId>,
77}
78
79/// Why a `meta.properties` does not read.
80///
81/// Each message is the message of the exception that Kafka's
82/// `PropertiesUtils.readPropertiesFile` and `MetaProperties.Builder` throw
83/// for the same file, except where a variant says otherwise.
84#[derive(Debug, Error)]
85#[non_exhaustive]
86pub enum MetaPropertiesError {
87    /// The file is there, but it cannot be read.
88    #[error(transparent)]
89    Io(#[from] io::Error),
90
91    /// A `\u` escape is not followed by four hexadecimal digits.
92    #[error("Malformed \\uxxxx encoding.")]
93    MalformedEscape,
94
95    /// `version` is not a number.
96    #[error("Invalid meta.properties version string '{0}'")]
97    InvalidVersion(String),
98
99    /// `version` is a number that Kafka does not define.
100    #[error("Unknown meta.properties version number {0}")]
101    UnknownVersion(i32),
102
103    /// The file is version 0, which only a `ZooKeeper` broker writes. Kafka
104    /// reads it. krabka does not: it is a krabka message.
105    #[error(
106        "Unsupported meta.properties version 0: krabka reads version 1, which a KRaft node writes"
107    )]
108    VersionZero,
109
110    /// The file has no `node.id`.
111    #[error("Failed to find node.id")]
112    NoNodeId,
113
114    /// `node.id` is not an `int`.
115    #[error("Unable to read node.id as a base-10 number.")]
116    NodeIdNotANumber,
117
118    /// The file has no `cluster.id`.
119    #[error("cluster.id was not found.")]
120    NoClusterId,
121
122    /// `cluster.id` is not a Kafka `Uuid`. Kafka accepts any string there, so
123    /// this is a krabka message, in the form of Kafka's message for
124    /// `directory.id`.
125    #[error("Unable to read cluster.id as a Uuid: {0}")]
126    ClusterIdNotAUuid(KafkaUuidError),
127
128    /// `directory.id` is not a Kafka `Uuid`.
129    #[error("Unable to read directory.id as a Uuid: {0}")]
130    DirectoryIdNotAUuid(KafkaUuidError),
131}
132
133impl MetaProperties {
134    /// Reads `<dir>/meta.properties`. Returns `None` when there is no such
135    /// file, which is how Kafka's `MetaPropertiesEnsemble.Loader` finds an
136    /// empty directory.
137    ///
138    /// # Errors
139    ///
140    /// Returns [`MetaPropertiesError::Io`] when the file is there but cannot
141    /// be read, and another variant when it does not parse.
142    pub fn read(dir: &Path) -> Result<Option<Self>, MetaPropertiesError> {
143        match fs::read(dir.join(META_PROPERTIES)) {
144            Ok(bytes) => Self::parse(&bytes).map(Some),
145            Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
146            Err(error) => Err(error.into()),
147        }
148    }
149
150    /// Parses the bytes of a `meta.properties`, as Kafka's
151    /// `new MetaProperties.Builder(props).build()` does after
152    /// `Properties.load`.
153    ///
154    /// The checks run in Kafka's order, so a file with several faults is
155    /// reported with Kafka's first message.
156    ///
157    /// # Errors
158    ///
159    /// Returns the [`MetaPropertiesError`] for the first fault.
160    pub fn parse(bytes: &[u8]) -> Result<Self, MetaPropertiesError> {
161        let props = load_properties(bytes)?;
162        let version = props.get(VERSION_PROP).map_or("0", String::as_str);
163        // `MetaPropertiesVersion.fromNumberString` trims as `String.trim`
164        // does, and `Integer.parseInt` does the rest.
165        let number: i32 = version
166            .trim_matches(|c: char| c <= ' ')
167            .parse()
168            .map_err(|_| MetaPropertiesError::InvalidVersion(version.to_owned()))?;
169        match number {
170            META_PROPERTIES_VERSION => {}
171            0 => return Err(MetaPropertiesError::VersionZero),
172            other => return Err(MetaPropertiesError::UnknownVersion(other)),
173        }
174        let node_id: i32 = props
175            .get(NODE_ID_PROP)
176            .ok_or(MetaPropertiesError::NoNodeId)?
177            .parse()
178            .map_err(|_| MetaPropertiesError::NodeIdNotANumber)?;
179        let cluster_id = props.get(CLUSTER_ID_PROP);
180        let directory_id: Option<DirectoryId> = props
181            .get(DIRECTORY_ID_PROP)
182            .map(|id| id.parse().map_err(MetaPropertiesError::DirectoryIdNotAUuid))
183            .transpose()?;
184        let cluster_id: ClusterId = cluster_id
185            .ok_or(MetaPropertiesError::NoClusterId)?
186            .parse()
187            .map_err(MetaPropertiesError::ClusterIdNotAUuid)?;
188        Ok(Self {
189            cluster_id,
190            node_id,
191            directory_id,
192        })
193    }
194
195    /// The text of the file that Kafka's `writePropertiesFile` writes for
196    /// these ids at `now`, with `now` in UTC.
197    ///
198    /// No value needs an escape: a node id is digits and a sign, and Kafka's
199    /// `Uuid` form uses only letters, digits, `-` and `_`, which
200    /// `Properties.store` writes as they are.
201    #[must_use]
202    pub fn to_file_text(&self, now: SystemTime) -> String {
203        let directory_id = self
204            .directory_id
205            .map_or_else(String::new, |id| format!("{DIRECTORY_ID_PROP}={id}\n"));
206        format!(
207            "#\n#{}\n{CLUSTER_ID_PROP}={}\n{directory_id}{NODE_ID_PROP}={}\n\
208             {VERSION_PROP}={META_PROPERTIES_VERSION}\n",
209            java_date(now),
210            self.cluster_id,
211            self.node_id,
212        )
213    }
214
215    /// Writes `<dir>/meta.properties` as Kafka's `writePropertiesFile` does
216    /// with `fsync` set: the text goes to `meta.properties.tmp`, which is
217    /// synced and renamed into place, and then, on Unix, the directory is
218    /// synced. A write that stops partway therefore never leaves a truncated
219    /// file.
220    ///
221    /// # Errors
222    ///
223    /// Returns the first I/O error. A rename that fails removes the
224    /// temporary file, as Kafka does.
225    pub fn write(&self, dir: &Path) -> io::Result<()> { self.write_observed(dir, |_| Ok(())) }
226
227    /// [`Self::write`], calling `after` with the path of each file once it
228    /// is written: the temporary file, then `meta.properties`. An error from
229    /// `after` stops the write there, which is how `krabka-format` injects a
230    /// fault for its tests.
231    pub(crate) fn write_observed(
232        &self,
233        dir: &Path,
234        mut after: impl FnMut(&Path) -> io::Result<()>,
235    ) -> io::Result<()> {
236        let text = self.to_file_text(SystemTime::now());
237        let tmp = dir.join(META_PROPERTIES_TMP);
238        let mut file = fs::File::create(&tmp)?;
239        file.write_all(text.as_bytes())?;
240        file.sync_all()?;
241        drop(file);
242        after(&tmp)?;
243        let path = dir.join(META_PROPERTIES);
244        if let Err(error) = fs::rename(&tmp, &path) {
245            // Kafka deletes the temporary file and reports the rename.
246            let _ = fs::remove_file(&tmp);
247            return Err(error);
248        }
249        // Syncing the directory is what makes the rename durable, and only
250        // Unix supports it through `std`: elsewhere, `wasm32-wasip1`
251        // included, opening or syncing a directory fails. The sync is
252        // skipped there, as `krabka-log` skips its own directory syncs.
253        #[cfg(unix)]
254        fs::File::open(dir)?.sync_all()?;
255        after(&path)
256    }
257}
258
259#[bon::builder]
260pub fn node_id_mismatch_message(stored: i32, previous: i32, path: &Path) -> String {
261    format!(
262        "Stored node id {stored} doesn't match previous node id {previous} in {}. If you \
263         moved your data, make sure your configured node id matches. If you intend \
264         to create a new node, you should remove all data in your data directories.",
265        path.display(),
266    )
267}
268
269/// Kafka's `MetaPropertiesEnsemble.verify`, over the readable files of the
270/// directories of one node, sorted by path.
271///
272/// The set agrees on one cluster id and one node id, and no two directories
273/// share a directory id, and no directory id is reserved. `expected_cluster_id`
274/// and `expected_node_id` are the ids that the set has to agree with, when
275/// they are known. Returns the cluster id of the set: `expected_cluster_id`
276/// when given, else the id of the first file, or `None` for an empty set.
277///
278/// # Errors
279///
280/// Returns Kafka's message, word for word, for the first disagreement.
281pub fn verify_ensemble<'a>(
282    formatted: impl IntoIterator<Item = (&'a Path, &'a MetaProperties)>,
283    expected_cluster_id: Option<ClusterId>,
284    expected_node_id: Option<i32>,
285) -> Result<Option<ClusterId>, String> {
286    let mut cluster_id = expected_cluster_id;
287    let mut node_id = expected_node_id;
288    let mut seen: HashMap<DirectoryId, &Path> = HashMap::new();
289    for (dir, meta) in formatted {
290        let path = dir.join(META_PROPERTIES);
291        match cluster_id {
292            None => cluster_id = Some(meta.cluster_id),
293            Some(expected) if expected != meta.cluster_id => {
294                return Err(format!(
295                    "Invalid cluster.id in: {}. Expected {expected}, but read {}",
296                    path.display(),
297                    meta.cluster_id,
298                ));
299            }
300            Some(_) => {}
301        }
302        match node_id {
303            None => node_id = Some(meta.node_id),
304            Some(expected) if expected != meta.node_id => {
305                return Err(node_id_mismatch_message()
306                    .stored(meta.node_id)
307                    .previous(expected)
308                    .path(&path)
309                    .call());
310            }
311            Some(_) => {}
312        }
313        if let Some(directory_id) = meta.directory_id {
314            if directory_id.is_reserved() {
315                return Err(format!(
316                    "Invalid reserved directory ID {directory_id} found in {}",
317                    dir.display(),
318                ));
319            }
320            if let Some(previous) = seen.insert(directory_id, dir) {
321                // Kafka prints the `Optional` wrapper of the id here.
322                return Err(format!(
323                    "Duplicate directory ID Optional[{directory_id}] found. It was the ID of {}, \
324                     but also of {}",
325                    previous.display(),
326                    dir.display(),
327                ));
328            }
329        }
330    }
331    Ok(cluster_id)
332}
333
334/// Kafka's `Properties.load(InputStream)`: the key and value of each entry.
335/// A key that appears twice keeps its last value.
336///
337/// `load(InputStream)` reads ISO 8859-1, so each byte is the character of
338/// the same value. An escape `\uXXXX` gives a UTF-16 code unit.
339fn load_properties(bytes: &[u8]) -> Result<HashMap<String, String>, MetaPropertiesError> {
340    let mut props = HashMap::new();
341    for line in logical_lines(bytes) {
342        let RawEntry { key, value } = split_entry(&line);
343        props.insert(unescape(key)?, unescape(value)?);
344    }
345    Ok(props)
346}
347
348/// The white space that `Properties.load` skips.
349fn is_white_space(byte: u8) -> bool { matches!(byte, b' ' | b'\t' | b'\x0c') }
350
351fn is_line_end(byte: u8) -> bool { matches!(byte, b'\n' | b'\r') }
352
353/// The logical lines of a properties file, as `Properties.LineReader` reads
354/// them.
355///
356/// A natural line ends at `\n`, `\r`, or `\r\n`. White space at the start of
357/// a line is skipped, and a blank line is skipped. A line whose first other
358/// character is `#` or `!` is a comment. A line that ends in an odd number of
359/// backslashes goes on in the next line: the last backslash is dropped, and
360/// so is the white space at the start of the next line.
361fn logical_lines(input: &[u8]) -> Vec<Vec<u8>> {
362    let mut lines = Vec::new();
363    let mut line = Vec::new();
364    let mut skip_white_space = true;
365    let mut appended_line_begin = false;
366    let mut preceding_backslash = false;
367    let mut skip_lf = false;
368    let mut bytes = input.iter().copied();
369    while let Some(byte) = bytes.next() {
370        if skip_lf {
371            skip_lf = false;
372            if byte == b'\n' {
373                continue;
374            }
375        }
376        if skip_white_space {
377            if is_white_space(byte) || (!appended_line_begin && is_line_end(byte)) {
378                continue;
379            }
380            skip_white_space = false;
381            appended_line_begin = false;
382        }
383        if line.is_empty() && matches!(byte, b'#' | b'!') {
384            if !bytes.by_ref().any(is_line_end) {
385                break;
386            }
387            skip_white_space = true;
388            continue;
389        }
390        if !is_line_end(byte) {
391            line.push(byte);
392            preceding_backslash = byte == b'\\' && !preceding_backslash;
393        } else if line.is_empty() {
394            skip_white_space = true;
395        } else if preceding_backslash {
396            line.pop();
397            skip_white_space = true;
398            appended_line_begin = true;
399            preceding_backslash = false;
400            skip_lf = byte == b'\r';
401        } else {
402            lines.push(mem::take(&mut line));
403            skip_white_space = true;
404        }
405    }
406    if !line.is_empty() {
407        if preceding_backslash {
408            line.pop();
409        }
410        lines.push(line);
411    }
412    lines
413}
414
415/// The raw key and value of one logical line, before their escapes resolve.
416struct RawEntry<'a> {
417    key: &'a [u8],
418    value: &'a [u8],
419}
420
421/// Splits a logical line into its key and its value, as `Properties.load0`
422/// does.
423///
424/// The key ends at the first `=`, `:`, or white space that no backslash
425/// escapes. The value starts after the white space that follows, with at
426/// most one `=` or `:` among it.
427fn split_entry(line: &[u8]) -> RawEntry<'_> {
428    let mut key_len = 0;
429    let mut value_start = line.len();
430    let mut has_separator = false;
431    let mut preceding_backslash = false;
432    while let Some(&byte) = line.get(key_len) {
433        if !preceding_backslash && matches!(byte, b'=' | b':') {
434            value_start = key_len + 1;
435            has_separator = true;
436            break;
437        }
438        if !preceding_backslash && is_white_space(byte) {
439            value_start = key_len + 1;
440            break;
441        }
442        preceding_backslash = byte == b'\\' && !preceding_backslash;
443        key_len += 1;
444    }
445    while let Some(&byte) = line.get(value_start) {
446        if !is_white_space(byte) {
447            if !has_separator && matches!(byte, b'=' | b':') {
448                has_separator = true;
449            } else {
450                break;
451            }
452        }
453        value_start += 1;
454    }
455    // Both scans stop inside `line`, so neither bound passes its end.
456    RawEntry {
457        key: line.get(..key_len).unwrap_or(line),
458        value: line.get(value_start..).unwrap_or_default(),
459    }
460}
461
462/// Resolves the escapes of a key or a value, as `Properties.loadConvert`
463/// does: `\t`, `\r`, `\n`, `\f`, and `\uXXXX` are the characters they name,
464/// and a backslash before any other character is dropped.
465fn unescape(raw: &[u8]) -> Result<String, MetaPropertiesError> {
466    let mut units: Vec<u16> = Vec::with_capacity(raw.len());
467    let mut bytes = raw.iter().copied();
468    while let Some(byte) = bytes.next() {
469        if byte != b'\\' {
470            units.push(u16::from(byte));
471            continue;
472        }
473        let Some(escaped) = bytes.next() else {
474            break;
475        };
476        let unit = match escaped {
477            b'u' => {
478                let mut value = 0u16;
479                for _ in 0..4 {
480                    let digit = bytes
481                        .next()
482                        .and_then(|digit| char::from(digit).to_digit(16))
483                        .and_then(|digit| u16::try_from(digit).ok())
484                        .ok_or(MetaPropertiesError::MalformedEscape)?;
485                    value = (value << 4) | digit;
486                }
487                value
488            }
489            b't' => u16::from(b'\t'),
490            b'r' => u16::from(b'\r'),
491            b'n' => u16::from(b'\n'),
492            b'f' => 0x0c,
493            other => u16::from(other),
494        };
495        units.push(unit);
496    }
497    Ok(String::from_utf16_lossy(&units))
498}
499
500/// `java.util.Date.toString` in UTC: `EEE MMM dd HH:mm:ss zzz yyyy`.
501///
502/// `Properties.store` writes it as the comment on the second line. A time
503/// before 1970 is written as 1970, which no clock that formats a directory
504/// reads.
505fn java_date(now: SystemTime) -> String {
506    let date = OffsetDateTime::from(now.max(UNIX_EPOCH));
507    // `EEE` and `MMM`: the first three letters of the English name.
508    let short = |name: String| name.chars().take(3).collect::<String>();
509    format!(
510        "{} {} {:02} {:02}:{:02}:{:02} UTC {}",
511        short(date.weekday().to_string()),
512        short(date.month().to_string()),
513        date.day(),
514        date.hour(),
515        date.minute(),
516        date.second(),
517        date.year(),
518    )
519}
520
521#[cfg(test)]
522mod tests {
523    use {super::*, std::time::Duration, uuid::Uuid};
524
525    const CLUSTER: &str = "AQIDBAUGBwgJCgsMDQ4PEA";
526    const DIRECTORY: &str = "U-hCdTzaSPyFB3swwc7Qdw";
527
528    /// A `verify_ensemble` case: what it is, the set, the expected cluster
529    /// id and node id, and the result.
530    struct EnsembleCase {
531        what: &'static str,
532        set: Vec<DirectoryFile>,
533        cluster_id: Option<ClusterId>,
534        node_id: Option<i32>,
535        want: Result<Option<ClusterId>, String>,
536    }
537
538    /// The `meta.properties` read from one directory.
539    struct DirectoryFile {
540        dir: &'static str,
541        meta: MetaProperties,
542    }
543
544    fn directory(dir: &'static str, meta: MetaProperties) -> DirectoryFile {
545        DirectoryFile { dir, meta }
546    }
547
548    fn at(seconds: u64) -> SystemTime { UNIX_EPOCH + Duration::from_secs(seconds) }
549
550    fn meta(node_id: i32, directory_id: Option<&str>) -> MetaProperties {
551        MetaProperties {
552            cluster_id: CLUSTER.parse().expect("cluster id"),
553            node_id,
554            directory_id: directory_id.map(|id| id.parse().expect("directory id")),
555        }
556    }
557
558    /// `Date.toString` in UTC, checked against `date -u` for each instant.
559    #[yare::parameterized(
560        epoch = { 0, "Thu Jan 01 00:00:00 UTC 1970" },
561        leap_day_2000 = { 951_782_400, "Tue Feb 29 00:00:00 UTC 2000" },
562        kafka_epoch_example = { 1_000_000_000, "Sun Sep 09 01:46:40 UTC 2001" },
563        leap_day_2024 = { 1_709_210_096, "Thu Feb 29 12:34:56 UTC 2024" },
564        year_2099_end = { 4_102_444_799, "Thu Dec 31 23:59:59 UTC 2099" },
565        zero_padded = { 1_767_596_889, "Mon Jan 05 07:08:09 UTC 2026" },
566    )]
567    fn the_date_comment_is_java_date_to_string(seconds: u64, want: &str) {
568        assert2::check!(java_date(at(seconds)) == want, "{seconds}");
569    }
570
571    #[test]
572    fn pre_epoch_date_clamps_to_epoch() {
573        assert2::check!(
574            java_date(UNIX_EPOCH - Duration::from_secs(1)) == "Thu Jan 01 00:00:00 UTC 1970"
575        );
576    }
577
578    /// The bytes are the bytes Kafka's `PropertiesUtils.writePropertiesFile`
579    /// writes for the same ids at the same time on a UTC host:
580    /// `Properties.store(writer, "")` puts the empty comment, the date
581    /// comment, and the entries in key order, each line ended by `\n`.
582    #[yare::parameterized(
583        with_directory_id = {
584            "a formatted directory",
585            meta(1, Some(DIRECTORY)),
586            "#\n#Thu Feb 29 12:34:56 UTC 2024\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n\
587             directory.id=U-hCdTzaSPyFB3swwc7Qdw\nnode.id=1\nversion=1\n"
588        },
589        without_directory_id = {
590            "no directory id",
591            meta(2_147_483_647, None),
592            "#\n#Thu Feb 29 12:34:56 UTC 2024\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n\
593             node.id=2147483647\nversion=1\n"
594        },
595    )]
596    fn the_file_is_the_one_kafka_writes(what: &str, meta: MetaProperties, want: &str) {
597        assert2::check!(meta.to_file_text(at(1_709_210_096)) == want, "{what}");
598    }
599
600    /// Every file krabka writes reads back as the same ids.
601    #[yare::parameterized(
602        ordinary = { meta(0, Some(DIRECTORY)) },
603        reserved_directory_id = { meta(1, Some("AAAAAAAAAAAAAAAAAAAAZA")) },
604        high_bit_directory_id = { meta(2_147_483_647, Some("_____________________w")) },
605        no_directory_id = { meta(7, None) },
606        random_directory_id = { MetaProperties {
607                cluster_id: ClusterId(Uuid::from_u128(0)),
608                node_id: 3,
609                directory_id: Some(DirectoryId::random()),
610            } },
611    )]
612    fn the_written_file_parses_back_to_the_same_ids(written: MetaProperties) {
613        let tmp = tempfile::tempdir().expect("tempdir");
614        written.write(tmp.path()).expect("write");
615        assert2::check!(MetaProperties::read(tmp.path()).expect("read") == Some(written));
616        assert2::check!(!tmp.path().join(META_PROPERTIES_TMP).exists());
617    }
618
619    #[test]
620    fn an_empty_directory_has_no_meta_properties() {
621        let empty = tempfile::tempdir().expect("tempdir");
622        assert2::check!(MetaProperties::read(empty.path()).expect("read").is_none());
623    }
624
625    /// `Properties.load` reads more than `store` writes: the file of an
626    /// older Java, `:` and white space as separators, comments, line
627    /// continuations, escapes, and `\r\n`. A key given twice keeps its last
628    /// value.
629    #[yare::parameterized(
630        kafka_hash_order = { "#\n#Thu Feb 29 12:34:56 UTC 2024\nnode.id=1\nversion=1\n\
631             directory.id=U-hCdTzaSPyFB3swwc7Qdw\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n" },
632        no_comments = { "version=1\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\nnode.id=1\n\
633             directory.id=U-hCdTzaSPyFB3swwc7Qdw\n" },
634        colon_and_whitespace_separators = { "version : 1\n  cluster.id:AQIDBAUGBwgJCgsMDQ4PEA\nnode.id 1\n\
635             \tdirectory.id \t= U-hCdTzaSPyFB3swwc7Qdw" },
636        comments_and_blank_lines = { "! a comment\n\n   # another\nversion=1\n\n\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n\
637             node.id=1\ndirectory.id=U-hCdTzaSPyFB3swwc7Qdw\n" },
638        crlf_and_cr_endings = { "#\r\nversion=1\r\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\rnode.id=1\r\n\
639             directory.id=U-hCdTzaSPyFB3swwc7Qdw\r\n" },
640        continued_line = { "version=1\ncluster.id=AQIDBAUGBwgJ\\\n    CgsMDQ4PEA\nnode.id=1\n\
641             directory.id=U-hCdTzaSPyFB3swwc7Qdw\\\r\n" },
642        escaped_key_and_value = { "vers\\ion=\\u0031\ncluster\\.id=AQIDBAUGBwgJCgsMDQ4PEA\nnode.id=1\n\
643             directory.id=U-hCdTzaSPyF\\B3swwc7Qdw\n" },
644        duplicate_key_keeps_last = { "version=1\nnode.id=9\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\nnode.id=1\n\
645             directory.id=U-hCdTzaSPyFB3swwc7Qdw\n" },
646        unknown_key_is_ignored = { "version=1\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\nnode.id=1\nbroker.id=4\n\
647             directory.id=U-hCdTzaSPyFB3swwc7Qdw\n" },
648    )]
649    fn the_reader_accepts_what_properties_load_accepts(text: &str) {
650        let want = meta(1, Some(DIRECTORY));
651        assert2::check!(
652            MetaProperties::parse(text.as_bytes()).map_err(|e| e.to_string()) == Ok(want),
653            "{text:?}"
654        );
655    }
656
657    /// A file that does not read gets the message of the exception Kafka
658    /// throws for it, and the first fault in Kafka's order wins.
659    #[yare::parameterized(
660        bad_version = { "version=one\n", "Invalid meta.properties version string 'one'" },
661        unknown_version = {
662            "version=2\nnode.id=1\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n",
663            "Unknown meta.properties version number 2",
664        },
665        unsupported_zero = {
666            "version=0\nbroker.id=1\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n",
667            "Unsupported meta.properties version 0: krabka reads version 1, which a KRaft node writes",
668        },
669        missing_version = {
670            "node.id=1\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n",
671            "Unsupported meta.properties version 0: krabka reads version 1, which a KRaft node writes",
672        },
673        old_json = {
674            "{\n  \"cluster_id\": \"AQIDBAUGBwgJCgsMDQ4PEA\",\n  \"version\": 3\n}\n",
675            "Unsupported meta.properties version 0: krabka reads version 1, which a KRaft node writes",
676        },
677        no_node_or_cluster = { "version=1\n", "Failed to find node.id" },
678        node_trailing_space = {
679            "version=1\nnode.id=1 \ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n",
680            "Unable to read node.id as a base-10 number.",
681        },
682        node_overflow = {
683            "version=1\nnode.id=2147483648\ncluster.id=AQIDBAUGBwgJCgsMDQ4PEA\n",
684            "Unable to read node.id as a base-10 number.",
685        },
686        bad_directory_precedes_cluster_missing = {
687            "version=1\nnode.id=1\ndirectory.id=nope\n",
688            "Unable to read directory.id as a Uuid: Input string `nope` decoded as 3 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
689        },
690        no_cluster = { "version=1\nnode.id=1\n", "cluster.id was not found." },
691        invalid_cluster = {
692            "version=1\nnode.id=1\ncluster.id=AAAAAA\n",
693            "Unable to read cluster.id as a Uuid: Input string `AAAAAA` decoded as 4 bytes, which is not equal to the expected 16 bytes of a base64-encoded UUID",
694        },
695        malformed_escape = {
696            "version=1\nnode.id=1\ncluster.id=\\u00zz\n",
697            "Malformed \\uxxxx encoding.",
698        },
699    )]
700    fn a_bad_file_gets_kafkas_message(text: &str, want: &str) {
701        assert2::check!(
702            MetaProperties::parse(text.as_bytes()).map_err(|e| e.to_string())
703                == Err(want.to_owned()),
704            "{text:?}"
705        );
706    }
707
708    /// `verify_ensemble` agrees with Kafka's `MetaPropertiesEnsemble.verify`
709    /// for each way a set can disagree.
710    #[yare::parameterized(
711        nothing_formatted = { "nothing formatted" },
712        one_cluster_and_node = { "one cluster, one node" },
713        set_supplies_node_id = { "the set supplies the node id" },
714        cluster_id_disagrees = { "the given cluster id disagrees" },
715        node_id_disagrees = { "the configured node id disagrees" },
716        directory_node_ids_disagree = { "two directories disagree on the node id" },
717        reserved_directory_id = { "a reserved directory id" },
718        shared_directory_id = { "a shared directory id" },
719        missing_directory_ids = { "no directory id is no conflict" },
720    )]
721    fn verify_ensemble_matches_kafka(scenario: &str) {
722        let c1 = ClusterId(Uuid::from_u128(0xc1));
723        let c2 = ClusterId(Uuid::from_u128(0xc2));
724        let file = |cluster_id, node_id, directory: u128| MetaProperties {
725            cluster_id,
726            node_id,
727            directory_id: Some(DirectoryId(Uuid::from_u128(directory))),
728        };
729        let cases: Vec<EnsembleCase> = vec![
730            EnsembleCase {
731                what: "nothing formatted",
732                set: vec![],
733                cluster_id: None,
734                node_id: Some(1),
735                want: Ok(None),
736            },
737            EnsembleCase {
738                what: "one cluster, one node",
739                set: vec![
740                    directory("/a", file(c1, 1, 0x1000)),
741                    directory("/b", file(c1, 1, 0x2000)),
742                ],
743                cluster_id: None,
744                node_id: Some(1),
745                want: Ok(Some(c1)),
746            },
747            EnsembleCase {
748                what: "the set supplies the node id",
749                set: vec![
750                    directory("/a", file(c1, 4, 0x1000)),
751                    directory("/b", file(c1, 4, 0x2000)),
752                ],
753                cluster_id: Some(c1),
754                node_id: None,
755                want: Ok(Some(c1)),
756            },
757            EnsembleCase {
758                what: "the given cluster id disagrees",
759                set: vec![directory("/a", file(c1, 1, 0x1000))],
760                cluster_id: Some(c2),
761                node_id: Some(1),
762                want: Err(format!(
763                    "Invalid cluster.id in: {}. Expected {c2}, but read {c1}",
764                    Path::new("/a").join(META_PROPERTIES).display()
765                )),
766            },
767            EnsembleCase {
768                what: "the configured node id disagrees",
769                set: vec![directory("/a", file(c1, 2, 0x1000))],
770                cluster_id: None,
771                node_id: Some(1),
772                want: Err(node_id_mismatch_message()
773                    .stored(2)
774                    .previous(1)
775                    .path(&Path::new("/a").join(META_PROPERTIES))
776                    .call()),
777            },
778            EnsembleCase {
779                what: "two directories disagree on the node id",
780                set: vec![
781                    directory("/a", file(c1, 3, 0x1000)),
782                    directory("/b", file(c1, 4, 0x2000)),
783                ],
784                cluster_id: None,
785                node_id: None,
786                want: Err(format!(
787                    "Stored node id 4 doesn't match previous node id 3 in {}. If you moved \
788                         your data, make sure your configured node id matches. If you intend to \
789                         create a new node, you should remove all data in your data directories.",
790                    Path::new("/b").join(META_PROPERTIES).display()
791                )),
792            },
793            EnsembleCase {
794                what: "a reserved directory id",
795                set: vec![directory("/a", file(c1, 1, 2))],
796                cluster_id: None,
797                node_id: Some(1),
798                want: Err(
799                    "Invalid reserved directory ID AAAAAAAAAAAAAAAAAAAAAg found in /a".to_owned(),
800                ),
801            },
802            EnsembleCase {
803                what: "a shared directory id",
804                set: vec![
805                    directory("/a", file(c1, 1, 0x1000)),
806                    directory("/b", file(c1, 1, 0x1000)),
807                ],
808                cluster_id: None,
809                node_id: Some(1),
810                want: Err(format!(
811                    "Duplicate directory ID Optional[{}] found. It was the ID of /a, but also of \
812                     /b",
813                    DirectoryId(Uuid::from_u128(0x1000))
814                )),
815            },
816            EnsembleCase {
817                what: "no directory id is no conflict",
818                set: vec![
819                    directory("/a", MetaProperties {
820                        directory_id: None,
821                        ..file(c1, 1, 0)
822                    }),
823                    directory("/b", MetaProperties {
824                        directory_id: None,
825                        ..file(c1, 1, 0)
826                    }),
827                ],
828                cluster_id: None,
829                node_id: Some(1),
830                want: Ok(Some(c1)),
831            },
832        ];
833        let EnsembleCase {
834            what,
835            set,
836            cluster_id,
837            node_id,
838            want,
839        } = cases
840            .into_iter()
841            .find(|case| case.what == scenario)
842            .expect("named yare case exists");
843        let got = verify_ensemble(
844            set.iter().map(|file| (Path::new(file.dir), &file.meta)),
845            cluster_id,
846            node_id,
847        );
848        assert2::check!(got == want, "{what}");
849    }
850}
krabka_format/meta_properties.rs

krabka_format/
meta_properties.rs
