krabka_log/
name.rs

1//! Names inside a Kafka log directory: the per-partition directory, and the segment files in it.
2//!
3//! A partition's data lives in `<log_dir>/<topic>-<partition>/`. Kafka names each segment in that directory by its 20-digit zero-padded base offset, with the `.log`, `.index`, and `.timeindex` extensions.
4//!
5//! The module is public because a tool that opens a log directory needs the same paths the broker uses, and it must not have to link the broker to compute them.
6
7use {
8    crate::error::LogError,
9    std::path::{Path, PathBuf},
10};
11
12/// Digit count of a segment filename stem. Kafka zero-pads the base offset to this width.
13pub const FILENAME_DIGITS: usize = 20;
14
15/// `0` → `"00000000000000000000"`. `1847` → `"00000000000000001847"`.
16#[must_use]
17pub fn format_base_offset(base_offset: i64) -> String { format!("{base_offset:020}") }
18
19/// Parse a `.log` filename and return its base offset.
20/// `"00000000000000001847.log"` → `Ok(1847)`.
21///
22/// # Errors
23///
24/// Returns [`LogError::BadSegmentName`] when the name has no `.log` extension, when the stem is not [`FILENAME_DIGITS`] characters long, or when the stem is not a decimal `i64`.
25pub fn parse_log_filename(name: &str) -> Result<i64, LogError> {
26    let stem = name
27        .strip_suffix(".log")
28        .ok_or_else(|| LogError::BadSegmentName(name.into()))?;
29    if stem.len() != FILENAME_DIGITS {
30        return Err(LogError::BadSegmentName(name.into()));
31    }
32    stem.parse::<i64>()
33        .map_err(|_| LogError::BadSegmentName(name.into()))
34}
35
36/// Path to the `.log` file of the segment that starts at `base_offset`. It holds the record batches.
37#[must_use]
38pub fn log_path(dir: &Path, base_offset: i64) -> PathBuf {
39    dir.join(format!("{}.log", format_base_offset(base_offset)))
40}
41
42/// Path to the sparse offset index, `.index`, of the segment that starts at `base_offset`.
43#[must_use]
44pub fn index_path(dir: &Path, base_offset: i64) -> PathBuf {
45    dir.join(format!("{}.index", format_base_offset(base_offset)))
46}
47
48/// Path to the sparse timestamp index, `.timeindex`, of the segment that starts at `base_offset`.
49#[must_use]
50pub fn timeindex_path(dir: &Path, base_offset: i64) -> PathBuf {
51    dir.join(format!("{}.timeindex", format_base_offset(base_offset)))
52}
53
54/// Path to the aborted-transaction index, `.txnindex`, of the segment that starts at `base_offset`.
55#[must_use]
56pub fn txnindex_path(dir: &Path, base_offset: i64) -> PathBuf {
57    dir.join(format!("{}.txnindex", format_base_offset(base_offset)))
58}
59
60/// Path to the `.snapshot` file that holds the producer state as of `offset`.
61#[must_use]
62pub fn producer_snapshot_path(dir: &Path, offset: i64) -> PathBuf {
63    dir.join(format!("{}.snapshot", format_base_offset(offset)))
64}
65
66/// Path to the per-segment `.stampindex` sidecar. It holds the additional
67/// internal stamp coordinate and is never a client-facing file.
68#[must_use]
69pub fn stampindex_path(dir: &Path, base_offset: i64) -> PathBuf {
70    dir.join(format!("{}.stampindex", format_base_offset(base_offset)))
71}
72
73/// Path to the per-partition `.leader-epoch-checkpoint` file.
74#[must_use]
75pub fn leader_epoch_checkpoint_path(dir: &Path) -> PathBuf { dir.join("leader-epoch-checkpoint") }
76
77/// Path to the per-partition `log-start-offset-checkpoint` file.
78///
79/// It holds the log start offset that no segment name records: the part of a
80/// trim that lands inside a segment rather than on a segment boundary.
81#[must_use]
82pub fn log_start_offset_checkpoint_path(dir: &Path) -> PathBuf {
83    dir.join("log-start-offset-checkpoint")
84}
85
86/// Suffix that the broker appends to a future-log partition directory while a KIP-113 intra-broker move runs.
87///
88/// The directory at `<target_log_dir>/<topic>-<partition><FUTURE_SUFFIX>` collects copied batches. When the future log catches up, the broker renames it in place to `<topic>-<partition>`.
89///
90/// This is not the name that Apache Kafka gives the directory. Kafka writes `<topic>-<partition>.<32-hex-uuid>-future` (`LocalLog.logFutureDirName`, `LogFileUtils.FUTURE_DIR_SUFFIX`), and a JVM broker cannot parse the name without the unique id. A partition has at most one move, and only the krabka broker reads the name, so the unique id is left out. `docs/format-divergences.md` lists the difference.
91pub const FUTURE_SUFFIX: &str = "-future";
92
93/// Builds the directory path for a (topic, partition).
94#[must_use]
95pub fn partition_dir(log_dir: &Path, topic: &str, partition: i32) -> PathBuf {
96    log_dir.join(format!("{topic}-{partition}"))
97}
98
99/// Parses `<topic>-<partition>` from a directory name. It returns `None` when the name does not match the pattern.
100#[must_use]
101pub fn parse_partition_dir(name: &str) -> Option<(String, i32)> {
102    let (topic, part) = name.rsplit_once('-')?;
103    if topic.is_empty() || topic.ends_with('-') {
104        // Empty topic or trailing `-` in the topic indicates a malformed
105        // name like "-0" or "foo--1" (which would otherwise parse the
106        // tail as a positive partition number).
107        return None;
108    }
109    let partition = part.parse::<i32>().ok()?;
110    if partition < 0 {
111        return None;
112    }
113    Some((topic.to_owned(), partition))
114}
115
116#[cfg(test)]
117mod tests {
118
119    use {super::*, assert2::assert};
120
121    fn check_offset_case(offset: i64, expected_filename: &str) {
122        let formatted = format_base_offset(offset);
123        assert!(formatted == expected_filename);
124        let parsed = parse_log_filename(&format!("{formatted}.log")).unwrap();
125        assert!(parsed == offset);
126    }
127
128    #[test]
129    fn zero() { check_offset_case(0, "00000000000000000000"); }
130    #[test]
131    fn small() { check_offset_case(1847, "00000000000000001847"); }
132    // Plan had "00000000001000000000000" (23 chars) which is wrong:
133    // `{:020}` zero-pads to 20 chars, and 1_000_000_000_000 is 13 digits,
134    // so we expect 7 leading zeros + "1000000000000" = 20 chars total.
135    #[test]
136    fn large() { check_offset_case(1_000_000_000_000, "00000001000000000000"); }
137
138    #[test]
139    fn rejects_non_log_extension() {
140        assert2::assert!(parse_log_filename("00000000000000000000.index").is_err());
141    }
142
143    #[yare::parameterized(
144        too_few_digits = { "123.log" },
145        too_many_digits = { "000000000000000001847.log" },
146    )]
147    fn rejects_wrong_digit_count(filename: &str) {
148        assert2::assert!(parse_log_filename(filename).is_err());
149    }
150
151    #[test]
152    fn round_trip_partition_dir() {
153        let p = partition_dir(Path::new("/tmp"), "foo", 7);
154        let name = p
155            .file_name()
156            .expect("path has a file name")
157            .to_str()
158            .expect("file name is utf-8");
159        assert!(parse_partition_dir(name) == Some(("foo".to_owned(), 7)));
160        assert!(parse_partition_dir("foo-0") == Some(("foo".to_owned(), 0)));
161    }
162
163    #[test]
164    fn rejects_negative_partition() {
165        assert!(parse_partition_dir("foo--1") == None);
166    }
167
168    #[test]
169    fn rejects_empty_topic() {
170        assert!(parse_partition_dir("-0") == None);
171    }
172
173    #[test]
174    fn rejects_no_dash() {
175        assert!(parse_partition_dir("foo") == None);
176    }
177
178    #[test]
179    fn handles_topic_with_dashes() {
180        // Topic names can themselves contain hyphens; rsplit takes the last.
181        assert!(parse_partition_dir("my-cool-topic-3") == Some(("my-cool-topic".to_owned(), 3)));
182    }
183}
krabka_log/name.rs

krabka_log/
name.rs
