krabka_metadata/
kafka_record.rs

1//! Bridge between [`MetadataRecord`] and a Kafka [`Record`] whose value is the
2//! wincode encoding of the record.
3//!
4//! Each [`MetadataRecord`] maps to exactly one [`Record`]: `key = None`,
5//! `value = wincode(MetadataRecord)`. The enum variant index is the record
6//! type, so no separate type tag is carried.
7//!
8//! Nothing in krabka writes this shape to the metadata log or sends it to an
9//! observer. The `__cluster_metadata` log, its snapshots, and the observer
10//! metadata fetch carry Kafka's KIP-631 `ApiMessageAndVersion` frame, which
11//! [`crate::kraft_translate`] writes and reads.
12
13use bytes::Bytes;
14use krabka_protocol::records::Record;
15use serde_wincode::SerdeCompat;
16use wincode::{Deserialize as _, Serialize as _};
17
18use crate::records::MetadataRecord;
19
20/// Error decoding a `Record` back into a [`MetadataRecord`].
21#[derive(Debug, thiserror::Error)]
22pub enum KafkaRecordError {
23    #[error("metadata record has no value payload")]
24    MissingValue,
25    #[error("wincode decode failed: {0}")]
26    Decode(String),
27    #[error("wincode encode failed: {0}")]
28    Encode(String),
29}
30
31/// Encode one [`MetadataRecord`] as a Kafka [`Record`].
32///
33/// # Errors
34/// Returns [`KafkaRecordError::Encode`] if wincode serialization fails. In
35/// practice this cannot happen for `MetadataRecord`.
36pub fn to_kafka_record(rec: &MetadataRecord) -> Result<Record, KafkaRecordError> {
37    let payload = <SerdeCompat<MetadataRecord>>::serialize(rec)
38        .map_err(|e| KafkaRecordError::Encode(e.to_string()))?;
39    Ok(Record {
40        value: Some(Bytes::from(payload)),
41        ..Default::default()
42    })
43}
44
45/// Decode a Kafka [`Record`] back into a [`MetadataRecord`].
46///
47/// # Errors
48/// - [`KafkaRecordError::MissingValue`] if the record carries no value.
49/// - [`KafkaRecordError::Decode`] if the value is not a valid
50///   wincode-encoded `MetadataRecord`.
51pub fn from_kafka_record(rec: &Record) -> Result<MetadataRecord, KafkaRecordError> {
52    let value = rec.value.as_ref().ok_or(KafkaRecordError::MissingValue)?;
53    <SerdeCompat<MetadataRecord>>::deserialize(value)
54        .map_err(|e| KafkaRecordError::Decode(e.to_string()))
55}
56
57#[cfg(test)]
58mod tests {
59
60    use uuid::Uuid;
61
62    use super::*;
63    use crate::records::{MetadataRecord, TopicRecord};
64
65    fn sample_topic() -> MetadataRecord {
66        MetadataRecord::V1Topic(TopicRecord {
67            name: "orders".into(),
68            topic_id: Uuid::from_u128(0x1234_5678_9abc_def0),
69            partitions: 6,
70            replication_factor: 3,
71        })
72    }
73
74    #[test]
75    fn round_trips_through_kafka_record() {
76        let rec = sample_topic();
77        let kafka = to_kafka_record(&rec).expect("encode");
78        let back = from_kafka_record(&kafka).expect("decode");
79        assert2::assert!(rec == back);
80    }
81
82    #[test]
83    fn key_is_none_and_value_is_present() {
84        let kafka = to_kafka_record(&sample_topic()).expect("encode");
85        assert2::assert!(kafka.key == None);
86        assert2::assert!(kafka.value.is_some());
87    }
88
89    #[test]
90    fn record_key_is_none_even_when_default_record_has_key() {
91        let kafka = to_kafka_record(&sample_topic()).expect("encode");
92        assert2::assert!(kafka.key == None);
93    }
94
95    #[test]
96    fn missing_value_is_an_error() {
97        let empty = Record::default();
98        assert2::assert!(matches!(
99            from_kafka_record(&empty),
100            Err(KafkaRecordError::MissingValue)
101        ));
102    }
103
104    #[test]
105    fn encoding_is_stable_across_calls() {
106        // Stability guard: the same record must always produce the same
107        // value bytes, so an observer fetching twice sees identical frames.
108        let rec = sample_topic();
109        let a = to_kafka_record(&rec).expect("encode");
110        let b = to_kafka_record(&rec).expect("encode");
111        assert2::assert!(a.value == b.value);
112    }
113}
krabka_metadata/kafka_record.rs

krabka_metadata/
kafka_record.rs
