kafka_meta/
snapshot.rs

1//! The owned metadata domain types.
2//!
3//! Every one of these is ours: rule 1 means no `kafka_protocol` type crosses
4//! this crate's public API, and metadata is where the temptation is strongest
5//! because `MetadataResponse` is *almost* the right shape. It is not quite —
6//! `BrokerId` and `TopicName` are newtypes, `StrBytes` is not `String`, `Uuid`
7//! comes from a crate we do not otherwise depend on, and all of it is
8//! `#[non_exhaustive]` and regenerated on every Kafka release.
9//!
10//! Snapshots are immutable and shared behind an `ArcSwap`. Readers never block
11//! and never see a half-updated cluster.
12
13use std::collections::HashMap;
14use std::fmt;
15use std::time::{Duration, Instant, SystemTime};
16
17use kafka_conn::ErrorCode;
18
19/// A Kafka topic id.
20///
21/// Ours rather than `uuid::Uuid`: it is a public field of [`TopicInfo`], and a
22/// dependency's type in a public struct is a dependency's semver in our
23/// semver.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
25pub struct TopicId([u8; 16]);
26
27impl TopicId {
28    /// All-zero, which is how the protocol spells "no id".
29    pub const ZERO: TopicId = TopicId([0; 16]);
30
31    /// Build from raw bytes.
32    pub const fn from_bytes(bytes: [u8; 16]) -> Self {
33        Self(bytes)
34    }
35
36    /// The raw bytes.
37    pub const fn as_bytes(&self) -> &[u8; 16] {
38        &self.0
39    }
40
41    /// Whether this is the zero id.
42    pub fn is_zero(&self) -> bool {
43        self.0 == [0; 16]
44    }
45}
46
47impl fmt::Display for TopicId {
48    /// Canonical 8-4-4-4-12 hex, the way `kafka-topics.sh` prints it.
49    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
50        for (index, byte) in self.0.iter().enumerate() {
51            if matches!(index, 4 | 6 | 8 | 10) {
52                f.write_str("-")?;
53            }
54            write!(f, "{byte:02x}")?;
55        }
56        Ok(())
57    }
58}
59
60/// A broker's advertised endpoint.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct BrokerInfo {
63    /// The broker id.
64    pub node_id: i32,
65    /// Advertised host.
66    pub host: String,
67    /// Advertised port.
68    pub port: i32,
69    /// Rack, when the broker declares one.
70    pub rack: Option<String>,
71}
72
73impl BrokerInfo {
74    /// `host:port`, as the connection layer wants it.
75    pub fn address(&self) -> String {
76        format!("{}:{}", self.host, self.port)
77    }
78}
79
80/// One partition of one topic.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct PartitionInfo {
83    /// Partition index.
84    pub partition: i32,
85    /// Current leader, or `None` when there is no leader right now.
86    ///
87    /// `None` rather than `-1`: a UI that renders "leader -1" is a UI that
88    /// forgot to check, and the type should not let it.
89    pub leader: Option<i32>,
90    /// Leader epoch, for fencing stale reads.
91    pub leader_epoch: i32,
92    /// The full replica set.
93    pub replicas: Vec<i32>,
94    /// In-sync replicas.
95    pub isr: Vec<i32>,
96    /// Replicas the leader considers offline.
97    pub offline_replicas: Vec<i32>,
98    /// A per-partition error, which does not invalidate the rest of the topic.
99    pub error: Option<ErrorCode>,
100}
101
102impl PartitionInfo {
103    /// Whether the partition has fewer in-sync replicas than replicas.
104    pub fn under_replicated(&self) -> bool {
105        self.isr.len() < self.replicas.len()
106    }
107}
108
109/// One topic.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct TopicInfo {
112    /// Topic name.
113    pub name: String,
114    /// Topic id, zero on brokers or versions that do not report one.
115    pub topic_id: TopicId,
116    /// Whether Kafka considers this an internal topic.
117    pub internal: bool,
118    /// Partitions, in index order.
119    pub partitions: Vec<PartitionInfo>,
120    /// A topic-level error — `UNKNOWN_TOPIC_OR_PARTITION` for a name that does
121    /// not exist, most often.
122    pub error: Option<ErrorCode>,
123}
124
125impl TopicInfo {
126    /// Look up one partition.
127    pub fn partition(&self, index: i32) -> Option<&PartitionInfo> {
128        self.partitions.iter().find(|p| p.partition == index)
129    }
130}
131
132/// An immutable view of the cluster.
133#[derive(Debug, Clone)]
134pub struct MetadataSnapshot {
135    brokers: Vec<BrokerInfo>,
136    brokers_by_id: HashMap<i32, usize>,
137    topics: Vec<TopicInfo>,
138    topics_by_name: HashMap<String, usize>,
139    controller_id: Option<i32>,
140    cluster_id: Option<String>,
141    fetched_at: SystemTime,
142    fetched_instant: Instant,
143}
144
145impl MetadataSnapshot {
146    /// Assemble a snapshot.
147    pub fn new(
148        brokers: Vec<BrokerInfo>,
149        topics: Vec<TopicInfo>,
150        controller_id: Option<i32>,
151        cluster_id: Option<String>,
152    ) -> Self {
153        let brokers_by_id = brokers
154            .iter()
155            .enumerate()
156            .map(|(index, broker)| (broker.node_id, index))
157            .collect();
158        let topics_by_name = topics
159            .iter()
160            .enumerate()
161            .map(|(index, topic)| (topic.name.clone(), index))
162            .collect();
163        Self {
164            brokers,
165            brokers_by_id,
166            topics,
167            topics_by_name,
168            controller_id,
169            cluster_id,
170            fetched_at: SystemTime::now(),
171            fetched_instant: Instant::now(),
172        }
173    }
174
175    /// An empty snapshot, for the moment before the first refresh lands.
176    pub fn empty() -> Self {
177        Self::new(Vec::new(), Vec::new(), None, None)
178    }
179
180    /// All known brokers.
181    pub fn brokers(&self) -> &[BrokerInfo] {
182        &self.brokers
183    }
184
185    /// One broker by id.
186    pub fn broker(&self, node_id: i32) -> Option<&BrokerInfo> {
187        self.brokers_by_id
188            .get(&node_id)
189            .and_then(|index| self.brokers.get(*index))
190    }
191
192    /// All known topics.
193    pub fn topics(&self) -> &[TopicInfo] {
194        &self.topics
195    }
196
197    /// One topic by name.
198    pub fn topic(&self, name: &str) -> Option<&TopicInfo> {
199        self.topics_by_name
200            .get(name)
201            .and_then(|index| self.topics.get(*index))
202    }
203
204    /// The active controller, when the broker told us.
205    pub fn controller_id(&self) -> Option<i32> {
206        self.controller_id
207    }
208
209    /// The cluster id.
210    pub fn cluster_id(&self) -> Option<&str> {
211        self.cluster_id.as_deref()
212    }
213
214    /// Wall-clock time this snapshot was fetched.
215    ///
216    /// A UI renders staleness; without this it can only render "now", which is
217    /// a lie whenever the refresh loop is stuck.
218    pub fn fetched_at(&self) -> SystemTime {
219        self.fetched_at
220    }
221
222    /// How long ago this snapshot was fetched, on a monotonic clock.
223    pub fn age(&self) -> Duration {
224        self.fetched_instant.elapsed()
225    }
226
227    /// The leader of one partition.
228    pub fn leader_for(&self, topic: &str, partition: i32) -> Option<i32> {
229        self.topic(topic)?.partition(partition)?.leader
230    }
231
232    /// Merge newer topic entries into this snapshot, keeping everything else.
233    ///
234    /// A targeted refresh asks about a handful of topics; discarding the rest
235    /// of the cache because of that would make every scan re-fetch the world.
236    pub fn with_topics_merged(&self, updated: Vec<TopicInfo>) -> Self {
237        let mut topics = self.topics.clone();
238        for topic in updated {
239            match self.topics_by_name.get(&topic.name) {
240                Some(index) => {
241                    if let Some(slot) = topics.get_mut(*index) {
242                        *slot = topic;
243                    }
244                }
245                None => topics.push(topic),
246            }
247        }
248        Self::new(
249            self.brokers.clone(),
250            topics,
251            self.controller_id,
252            self.cluster_id.clone(),
253        )
254    }
255}
256
257#[cfg(test)]
258mod tests {
259    use super::*;
260
261    fn broker(id: i32) -> BrokerInfo {
262        BrokerInfo {
263            node_id: id,
264            host: format!("broker-{id}"),
265            port: 9092,
266            rack: None,
267        }
268    }
269
270    fn topic(name: &str, leaders: &[i32]) -> TopicInfo {
271        TopicInfo {
272            name: name.to_owned(),
273            topic_id: TopicId::ZERO,
274            internal: false,
275            partitions: leaders
276                .iter()
277                .enumerate()
278                .map(|(index, leader)| PartitionInfo {
279                    partition: i32::try_from(index).unwrap_or(0),
280                    leader: Some(*leader),
281                    leader_epoch: 0,
282                    replicas: vec![*leader],
283                    isr: vec![*leader],
284                    offline_replicas: Vec::new(),
285                    error: None,
286                })
287                .collect(),
288            error: None,
289        }
290    }
291
292    #[test]
293    fn lookups_are_by_index_not_by_scan() {
294        let snapshot = MetadataSnapshot::new(
295            vec![broker(1), broker(2)],
296            vec![topic("orders", &[1, 2])],
297            Some(1),
298            Some("cluster".to_owned()),
299        );
300        assert_eq!(
301            snapshot.broker(2).map(|b| b.address()),
302            Some("broker-2:9092".to_owned())
303        );
304        assert!(snapshot.broker(99).is_none());
305        assert_eq!(snapshot.leader_for("orders", 1), Some(2));
306        assert_eq!(snapshot.leader_for("orders", 7), None);
307        assert_eq!(snapshot.leader_for("nope", 0), None);
308    }
309
310    #[test]
311    fn a_targeted_refresh_keeps_the_topics_it_did_not_ask_about() {
312        let snapshot = MetadataSnapshot::new(
313            vec![broker(1)],
314            vec![topic("orders", &[1]), topic("events", &[1])],
315            Some(1),
316            None,
317        );
318        let merged = snapshot.with_topics_merged(vec![topic("orders", &[1, 1, 1])]);
319        assert_eq!(merged.topics().len(), 2);
320        assert_eq!(merged.topic("orders").map(|t| t.partitions.len()), Some(3));
321        assert!(merged.topic("events").is_some());
322    }
323
324    #[test]
325    fn a_targeted_refresh_can_add_a_topic() {
326        let snapshot = MetadataSnapshot::new(vec![broker(1)], vec![], None, None);
327        let merged = snapshot.with_topics_merged(vec![topic("new", &[1])]);
328        assert!(merged.topic("new").is_some());
329    }
330
331    #[test]
332    fn a_missing_leader_is_none_not_minus_one() {
333        // The protocol spells "no leader" as -1. Letting that reach a UI is how
334        // you get a broker detail page for node -1.
335        let mut info = topic("orders", &[1]);
336        if let Some(p) = info.partitions.get_mut(0) {
337            p.leader = None;
338        }
339        let snapshot = MetadataSnapshot::new(vec![broker(1)], vec![info], None, None);
340        assert_eq!(snapshot.leader_for("orders", 0), None);
341    }
342
343    #[test]
344    fn topic_ids_render_as_uuids() {
345        let id = TopicId::from_bytes([
346            0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab,
347            0xcd, 0xef,
348        ]);
349        assert_eq!(id.to_string(), "01234567-89ab-cdef-0123-456789abcdef");
350        assert!(TopicId::ZERO.is_zero());
351        assert!(!id.is_zero());
352    }
353
354    #[test]
355    fn under_replicated_is_a_comparison_not_a_guess() {
356        let partition = PartitionInfo {
357            partition: 0,
358            leader: Some(1),
359            leader_epoch: 0,
360            replicas: vec![1, 2, 3],
361            isr: vec![1, 2],
362            offline_replicas: vec![3],
363            error: None,
364        };
365        assert!(partition.under_replicated());
366    }
367}
kafka_meta/snapshot.rs

kafka_meta/
snapshot.rs
